restructure and fix merge-conflicts

Author: skwowet

services/apps/script_executor_worker/src/activities.ts

@@ -1,13 +1,10 @@
 import { excludeEntityFromCleanup, hasActivityRecords } from './activities/cleanup/helpers'
 import { deleteMember, getMembersToCleanup } from './activities/cleanup/member'
-import {
-  deleteOrganization,
-  getOrganizationsToCleanup,
-  queueOrgForAggComputation,
-} from './activities/cleanup/organization'
+import { deleteOrganization, getOrganizationsToCleanup } from './activities/cleanup/organization'
 import {
   mergeMembers,
   mergeOrganizations,
+  queueOrganizationForAggComputation,
   unmergeMembers,
   unmergeMembersPreview,
   waitForTemporalWorkflowExecutionFinish,
@@ -25,13 +22,10 @@ import {
   findMemberMergeActions,
 } from './activities/dissect-member'
 import {
-  addOrgIdToRedisCache,
   calculateMemberAffiliations,
-  copyActivitiesFromPgToQuestDb,
   deleteProcessedMemberOrgAffiliations,
   getMembersWithDeletedOrgAffilations,
   getProcessedMemberOrgAffiliations,
-  hasActivityInQuestDb,
   markMemberOrgAffiliationAsProcessed,
 } from './activities/fix-deleted-member-org-affilations'
 import {
@@ -86,19 +80,16 @@ export {
   excludeEntityFromCleanup,
   getOrganizationsToCleanup,
   hasActivityRecords,
-  queueOrgForAggComputation,
   syncMembersBatch,
   getMembersForSync,
   getOrganizationsForSync,
   syncOrganizationsBatch,
   deleteIndexedEntities,
   markEntitiesIndexed,
+  queueOrganizationForAggComputation,
   getActivitiesToCopyToTinybird,
   markActivitiesAsIndexedForSyncingActivitiesToTinybird,
   getMembersWithDeletedOrgAffilations,
-  hasActivityInQuestDb,
-  copyActivitiesFromPgToQuestDb,
-  addOrgIdToRedisCache,
   calculateMemberAffiliations,
   markMemberOrgAffiliationAsProcessed,
   getProcessedMemberOrgAffiliations,

services/apps/script_executor_worker/src/activities/cleanup/organization.ts

@@ -21,12 +21,3 @@ export async function deleteOrganization(orgId: string): Promise<void> {
     throw error
   }
 }
-
-export async function queueOrgForAggComputation(orgId: string): Promise<void> {
-  try {
-    await svc.redis.sAdd('organizationIdsForAggComputation', orgId)
-  } catch (error) {
-    svc.log.error(error, 'Error adding organization to redis set!')
-    throw error
-  }
-}

services/apps/script_executor_worker/src/activities/common.ts

@@ -134,3 +134,12 @@ export function timeout(ms: number, workflowId: string): Promise<void> {
     }, ms)
   })
 }
+
+export async function queueOrganizationForAggComputation(orgId: string): Promise<void> {
+  try {
+    await svc.redis.sAdd('organizationIdsForAggComputation', orgId)
+  } catch (error) {
+    svc.log.error(error, 'Error adding organization to redis set!')
+    throw error
+  }
+}

services/apps/script_executor_worker/src/activities/fix-deleted-member-org-affilations/index.ts

@@ -1,7 +1,4 @@
-import { insertActivities } from '@crowd/data-access-layer'
-import { IDbActivity } from '@crowd/data-access-layer/src/old/apps/data_sink_worker/repo/activity.data'
 import { runMemberAffiliationsUpdate } from '@crowd/data-access-layer/src/old/apps/profiles_worker'
-import ActivityRepository from '@crowd/data-access-layer/src/old/apps/script_executor_worker/activity.repo'
 import MemberRepository from '@crowd/data-access-layer/src/old/apps/script_executor_worker/member.repo'
 import TempRepository from '@crowd/data-access-layer/src/old/apps/script_executor_worker/temp.repo'
 
@@ -18,36 +15,6 @@ export async function getMembersWithDeletedOrgAffilations(
   }
 }
 
-export async function hasActivityInQuestDb(
-  memberId: string,
-  organizationId: string,
-): Promise<boolean> {
-  const repo = new ActivityRepository(svc.postgres.reader.connection(), svc.log, svc.questdbSQL)
-  return repo.hasActivity(memberId, organizationId)
-}
-
-export async function copyActivitiesFromPgToQuestDb(
-  memberId: string,
-  orgId: string,
-): Promise<void> {
-  let offset = 0
-  let activities: IDbActivity[] = []
-
-  const repo = new ActivityRepository(svc.postgres.reader.connection(), svc.log, svc.questdbSQL)
-
-  // Get first batch
-  activities = await repo.findActivitiesPg(memberId, orgId, { offset: 0 })
-
-  while (activities.length > 0) {
-    // Insert current batch
-    await insertActivities(svc.queue, activities)
-
-    // Move to next batch
-    offset += activities.length
-    activities = await repo.findActivitiesPg(memberId, orgId, { offset })
-  }
-}
-
 export async function calculateMemberAffiliations(memberId: string): Promise<void> {
   try {
     await runMemberAffiliationsUpdate(svc.postgres.writer, svc.questdbSQL, svc.queue, memberId)
@@ -56,10 +23,6 @@ export async function calculateMemberAffiliations(memberId: string): Promise<voi
   }
 }
 
-export async function addOrgIdToRedisCache(orgId: string): Promise<void> {
-  await svc.redis.sAdd('organizationIdsForAggComputation', orgId)
-}
-
 export async function markMemberOrgAffiliationAsProcessed(
   memberId: string,
   organizationId: string,

services/apps/script_executor_worker/src/workflows.ts

@@ -7,8 +7,8 @@ import { findAndMergeMembersWithSameVerifiedEmailsInDifferentPlatforms } from '.
 import { fixDeletedMemberOrgAffilations } from './workflows/fixDeletedMemberOrgAffilations'
 import { fixOrgIdentitiesWithWrongUrls } from './workflows/fixOrgIdentitiesWithWrongUrls'
 import { populateActivityRelations } from './workflows/populateActivityRelations'
-import { syncMembers } from './workflows/syncMembers'
-import { syncOrganizations } from './workflows/syncOrganizations'
+import { syncMembers } from './workflows/sync/members'
+import { syncOrganizations } from './workflows/sync/organizations'
 
 export {
   findAndMergeMembersWithSameVerifiedEmailsInDifferentPlatforms,

services/apps/script_executor_worker/src/workflows/cleanup/organizations.ts

@@ -3,10 +3,11 @@ import { continueAsNew, proxyActivities } from '@temporalio/workflow'
 import { EntityType } from '@crowd/data-access-layer/src/old/apps/script_executor_worker/types'
 
 import * as cleanupHelpers from '../../activities/cleanup/helpers'
-import * as activities from '../../activities/cleanup/organization'
+import * as cleanupActivities from '../../activities/cleanup/organization'
+import * as commonActivities from '../../activities/common'
 import { ICleanupArgs } from '../../types'
 
-const activity = proxyActivities<typeof activities>({
+const cleanupActivity = proxyActivities<typeof cleanupActivities>({
   startToCloseTimeout: '30 minutes',
   retry: { maximumAttempts: 3, backoffCoefficient: 3 },
 })
@@ -16,10 +17,15 @@ const cleanupHelper = proxyActivities<typeof cleanupHelpers>({
   retry: { maximumAttempts: 3, backoffCoefficient: 3 },
 })
 
+const commonActivity = proxyActivities<typeof commonActivities>({
+  startToCloseTimeout: '30 minutes',
+  retry: { maximumAttempts: 3, backoffCoefficient: 3 },
+})
+
 export async function cleanupOrganizations(args: ICleanupArgs): Promise<void> {
   const BATCH_SIZE = args.batchSize ?? 100
 
-  const organizationIds = await activity.getOrganizationsToCleanup(BATCH_SIZE)
+  const organizationIds = await cleanupActivity.getOrganizationsToCleanup(BATCH_SIZE)
 
   if (organizationIds.length === 0) {
     console.log('No more organizations to cleanup!')
@@ -41,8 +47,8 @@ export async function cleanupOrganizations(args: ICleanupArgs): Promise<void> {
 
       console.log(`Deleting organization ${orgId} from database!`)
 
-      await activity.deleteOrganization(orgId)
-      return activity.queueOrgForAggComputation(orgId)
+      await cleanupActivity.deleteOrganization(orgId)
+      return commonActivity.queueOrganizationForAggComputation(orgId)
     })
 
     await Promise.all(cleanupTasks)

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

@@ -1,5 +1,6 @@
 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 { IFixDeletedMemberOrgAffilationsArgs } from '../types'
@@ -14,6 +15,11 @@ const syncActivity = proxyActivities<typeof syncActivities>({
   retry: { maximumAttempts: 3, backoffCoefficient: 3 },
 })
 
+const commonActivity = proxyActivities<typeof commonActivities>({
+  startToCloseTimeout: '30 minutes',
+  retry: { maximumAttempts: 3, backoffCoefficient: 3 },
+})
+
 export async function fixDeletedMemberOrgAffilations(
   args: IFixDeletedMemberOrgAffilationsArgs,
 ): Promise<void> {
@@ -23,7 +29,7 @@ export async function fixDeletedMemberOrgAffilations(
     `Fixing deleted member org affiliations with params: testRun=${args.testRun}, batchSize=${args.batchSize}`,
   )
 
-  // 1. Find affected memberId and orgId
+  // Find affected memberId and orgId
   const affectedMembers = await activity.getProcessedMemberOrgAffiliations(BATCH_SIZE)
 
   if (affectedMembers.length === 0) {
@@ -36,16 +42,6 @@ export async function fixDeletedMemberOrgAffilations(
     const chunk = affectedMembers.slice(i, i + CHUNK_SIZE)
     await Promise.all(
       chunk.map(async ({ memberId, organizationId }) => {
-        // 2. Check if they have activity in questDb
-        // const hasActivity = await activity.hasActivityInQuestDb(memberId, organizationId)
-
-        // 2.1 If no activities found, we need to get and insert them
-        // if (!hasActivity) {
-        //   console.log(`Copying activities for member ${memberId} and org ${organizationId}`)
-
-        //   await activity.copyActivitiesFromPgToQuestDb(memberId, organizationId)
-        // }
-
         // Calculate affiliation
         await activity.calculateMemberAffiliations(memberId)
 
@@ -54,7 +50,7 @@ export async function fixDeletedMemberOrgAffilations(
 
         // Add orgId to redisCache
         // It will be picked up by the spawnOrganizationAggregatesComputation workflow
-        await activity.addOrgIdToRedisCache(organizationId)
+        await commonActivity.queueOrganizationForAggComputation(organizationId)
 
         await activity.deleteProcessedMemberOrgAffiliations(memberId, organizationId)
       }),

…utor_worker/src/workflows/syncMembers.ts …tor_worker/src/workflows/sync/members.tsservices/apps/script_executor_worker/src/workflows/syncMembers.ts renamed to services/apps/script_executor_worker/src/workflows/sync/members.ts services/apps/script_executor_worker/src/workflows/syncMembers.ts renamed to services/apps/script_executor_worker/src/workflows/sync/members.ts

@@ -2,9 +2,9 @@ import { continueAsNew, proxyActivities } from '@temporalio/workflow'
 
 import { IndexedEntityType } from '@crowd/opensearch/src/repo/indexing.data'
 
-import * as entityIndexActivities from '../activities/sync/entity-index'
-import * as memberSyncActivities from '../activities/sync/member'
-import { ISyncArgs } from '../types'
+import * as entityIndexActivities from '../../activities/sync/entity-index'
+import * as memberSyncActivities from '../../activities/sync/member'
+import { ISyncArgs } from '../../types'
 
 const memberSyncActivity = proxyActivities<typeof memberSyncActivities>({
   startToCloseTimeout: '30 minutes',

…orker/src/workflows/syncOrganizations.ts …rker/src/workflows/sync/organizations.tsservices/apps/script_executor_worker/src/workflows/syncOrganizations.ts renamed to services/apps/script_executor_worker/src/workflows/sync/organizations.ts services/apps/script_executor_worker/src/workflows/syncOrganizations.ts renamed to services/apps/script_executor_worker/src/workflows/sync/organizations.ts

@@ -2,9 +2,9 @@ import { continueAsNew, proxyActivities } from '@temporalio/workflow'
 
 import { IndexedEntityType } from '@crowd/opensearch/src/repo/indexing.data'
 
-import * as entityIndexActivities from '../activities/sync/entity-index'
-import * as orgSyncActivities from '../activities/sync/organization'
-import { ISyncArgs } from '../types'
+import * as entityIndexActivities from '../../activities/sync/entity-index'
+import * as orgSyncActivities from '../../activities/sync/organization'
+import { ISyncArgs } from '../../types'
 
 const orgSyncActivity = proxyActivities<typeof orgSyncActivities>({
   startToCloseTimeout: '30 minutes',