chore: add script to block organization members affiliation (#3469)

Author: skwowet

services/apps/script_executor_worker/src/activities.ts

@@ -1,3 +1,7 @@
+import {
+  blockMemberOrganizationAffiliation,
+  getOrganizationMembers,
+} from './activities/block-organization-affiliation'
 import {
   findDuplicateMembersAfterDate,
   moveMemberActivityRelations,
@@ -124,4 +128,6 @@ export {
   checkActivitiesWithTimestampExistInQuestDb,
   saveMissingActivityInQuestDb,
   getMissingActivityInQuestDb,
+  blockMemberOrganizationAffiliation,
+  getOrganizationMembers,
 }

services/apps/script_executor_worker/src/activities/block-organization-affiliation.ts

@@ -0,0 +1,38 @@
+import { pgpQx } from '@crowd/data-access-layer'
+import { changeOverride } from '@crowd/data-access-layer/src/member_organization_affiliation_overrides'
+import OrganizationRepository from '@crowd/data-access-layer/src/old/apps/script_executor_worker/organization.repo'
+import { IMemberOrganization } from '@crowd/types'
+
+import { svc } from '../main'
+
+export async function getOrganizationMembers(
+  organizationId: string,
+  limit = 100,
+  offset = 0,
+): Promise<IMemberOrganization[]> {
+  try {
+    const orgRepo = new OrganizationRepository(svc.postgres.reader.connection(), svc.log)
+    return orgRepo.findOrganizationMembers(organizationId, limit, offset)
+  } catch (error) {
+    svc.log.error(error, 'Error getting organization members!')
+    throw error
+  }
+}
+
+export async function blockMemberOrganizationAffiliation(
+  memberId: string,
+  memberOrganizationId: string,
+): Promise<void> {
+  try {
+    const qx = pgpQx(svc.postgres.writer.connection())
+    return changeOverride(qx, {
+      memberId,
+      memberOrganizationId,
+      allowAffiliation: false,
+      isPrimaryWorkExperience: false,
+    })
+  } catch (error) {
+    svc.log.error(error, 'Error blocking organization affiliation!')
+    throw error
+  }
+}

services/apps/script_executor_worker/src/types.ts

@@ -63,3 +63,8 @@ export interface IDedupActivityRelationsArgs extends IScriptBatchTestArgs {
   groupsPerRun?: number
   cursor?: Omit<IActivityRelationDuplicateGroup, 'activityIds'>
 }
+
+export interface IBlockOrganizationAffiliationArgs {
+  organizationId: string
+  offset?: number
+}

services/apps/script_executor_worker/src/workflows.ts

@@ -1,3 +1,4 @@
+import { blockOrganizationAffiliation } from './workflows/block-organization-affiliation'
 import { cleanupDuplicateMembers } from './workflows/cleanup/duplicate-members'
 import { cleanupMembers } from './workflows/cleanup/members'
 import { cleanupOrganizations } from './workflows/cleanup/organizations'
@@ -30,4 +31,5 @@ export {
   cleanupDuplicateMembers,
   fixBotMembersAffiliation,
   dedupActivityRelations,
+  blockOrganizationAffiliation,
 }

services/apps/script_executor_worker/src/workflows/block-organization-affiliation.ts

@@ -0,0 +1,46 @@
+import { continueAsNew, proxyActivities } from '@temporalio/workflow'
+
+import * as activities from '../activities'
+import { IBlockOrganizationAffiliationArgs } from '../types'
+import { chunkArray } from '../utils/common'
+
+const { getOrganizationMembers, blockMemberOrganizationAffiliation, calculateMemberAffiliations } =
+  proxyActivities<typeof activities>({
+    startToCloseTimeout: '30 minutes',
+  })
+
+export async function blockOrganizationAffiliation(
+  args: IBlockOrganizationAffiliationArgs,
+): Promise<void> {
+  const MEMBERS_PER_RUN = 500
+  const BATCH_SIZE = 50
+  const OFFSET = args.offset ?? 0
+
+  const memberOrganizations = await getOrganizationMembers(
+    args.organizationId,
+    MEMBERS_PER_RUN,
+    OFFSET,
+  )
+
+  if (memberOrganizations.length === 0) {
+    console.log('No more organization members to block!')
+    return
+  }
+
+  // Step 1: Block all affiliations in batches
+  for (const chunk of chunkArray(memberOrganizations, BATCH_SIZE)) {
+    await Promise.all(chunk.map((mo) => blockMemberOrganizationAffiliation(mo.memberId, mo.id)))
+  }
+
+  // Step 2: Deduplicate memberIds and calculate affiliations
+  const uniqueMemberIds = Array.from(new Set(memberOrganizations.map((mo) => mo.memberId)))
+  for (const chunk of chunkArray(uniqueMemberIds, BATCH_SIZE)) {
+    await Promise.all(chunk.map((memberId) => calculateMemberAffiliations(memberId)))
+  }
+
+  // Step 3: Continue pagination
+  await continueAsNew<typeof blockOrganizationAffiliation>({
+    ...args,
+    offset: OFFSET + MEMBERS_PER_RUN,
+  })
+}

services/libs/data-access-layer/src/old/apps/script_executor_worker/organization.repo.ts

@@ -1,5 +1,6 @@
 import { DbConnection, DbTransaction } from '@crowd/database'
 import { Logger } from '@crowd/logging'
+import { IMemberOrganization } from '@crowd/types'
 
 class OrganizationRepository {
   constructor(
@@ -161,6 +162,21 @@ class OrganizationRepository {
       { memberId },
     )
   }
+
+  async findOrganizationMembers(
+    organizationId: string,
+    limit: number,
+    offset: number,
+  ): Promise<IMemberOrganization[]> {
+    return this.connection.any(
+      `
+      SELECT * FROM "memberOrganizations" 
+      WHERE "organizationId" = $(organizationId) AND "deletedAt" IS NULL 
+      ORDER BY "memberId", "id"
+      LIMIT $(limit) OFFSET $(offset)`,
+      { organizationId, limit, offset },
+    )
+  }
 }
 
 export default OrganizationRepository