chore: script to block project organization affiliations

Author: skwowet

services/apps/script_executor_worker/src/activities.ts

@@ -1,7 +1,9 @@
 import {
   blockMemberOrganizationAffiliation,
-  getOrganizationMembers,
-} from './activities/block-organization-affiliation'
+  fetchProjectMemberOrganizationsToBlock,
+  getMembersForAffiliationRecalc,
+  markMemberForAffiliationRecalc,
+} from './activities/block-project-organization-affiliations'
 import {
   findDuplicateMembersAfterDate,
   moveMemberActivityRelations,
@@ -78,6 +80,8 @@ export {
   unlinkOrganizationFromBotActivities,
   syncMember,
   blockMemberOrganizationAffiliation,
-  getOrganizationMembers,
+  fetchProjectMemberOrganizationsToBlock,
+  markMemberForAffiliationRecalc,
+  getMembersForAffiliationRecalc,
   calculateMemberAffiliations,
 }

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

@@ -0,0 +1,49 @@
+import { pgpQx } from '@crowd/data-access-layer'
+import { changeMemberOrganizationAffiliationOverrides } 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 { IChangeAffiliationOverrideData, IMemberOrganization } from '@crowd/types'
+
+import { svc } from '../main'
+
+export async function fetchProjectMemberOrganizationsToBlock(
+  limit: number,
+  afterId?: string,
+): Promise<Pick<IMemberOrganization, 'memberId' | 'id'>[]> {
+  try {
+    const orgRepo = new OrganizationRepository(svc.postgres.reader.connection(), svc.log)
+    return orgRepo.fetchProjectMemberOrganizationsToBlock(limit, afterId)
+  } catch (error) {
+    svc.log.error(error, 'Error fetching project member organizations to block!')
+    throw error
+  }
+}
+
+export async function blockMemberOrganizationAffiliation(
+  data: IChangeAffiliationOverrideData[],
+): Promise<void> {
+  try {
+    const qx = pgpQx(svc.postgres.writer.connection())
+    return changeMemberOrganizationAffiliationOverrides(qx, data)
+  } catch (error) {
+    svc.log.error(error, 'Error blocking organization affiliation!')
+    throw error
+  }
+}
+
+export async function markMemberForAffiliationRecalc(memberIds: string[]): Promise<void> {
+  try {
+    await svc.redis.sAdd('queue:recalculate:members:affiliation', memberIds)
+  } catch (error) {
+    svc.log.error(error, 'Error marking member for affiliation recalc!')
+    throw error
+  }
+}
+
+export async function getMembersForAffiliationRecalc(batchSize: number): Promise<string[]> {
+  try {
+    return svc.redis.sPop('queue:recalculate:members:affiliation', batchSize)
+  } catch (error) {
+    svc.log.error(error, 'Error getting members for affiliation recalc!')
+    throw error
+  }
+}

services/apps/script_executor_worker/src/activities/block-project-organization-affiliations.ts

@@ -1,7 +1,7 @@
 import MergeActionRepository from '@crowd/data-access-layer/src/old/apps/script_executor_worker/mergeAction.repo'
 import { EntityType } from '@crowd/data-access-layer/src/old/apps/script_executor_worker/types'
 
-import { svc } from '../../main'
+import { svc } from '../main'
 
 export async function getUnprocessedLLMApprovedSuggestions(
   batchSize: number,

…ies/process-llm-verified-merges/index.ts …ctivities/process-llm-verified-merges.tsservices/apps/script_executor_worker/src/activities/process-llm-verified-merges/index.ts renamed to services/apps/script_executor_worker/src/activities/process-llm-verified-merges.ts services/apps/script_executor_worker/src/activities/process-llm-verified-merges/index.ts renamed to services/apps/script_executor_worker/src/activities/process-llm-verified-merges.ts

@@ -50,6 +50,5 @@ export interface IDedupActivityRelationsArgs extends IScriptBatchTestArgs {
 }
 
 export interface IBlockOrganizationAffiliationArgs {
-  organizationId: string
-  offset?: number
+  afterId?: string
 }

services/apps/script_executor_worker/src/types.ts

@@ -1,4 +1,4 @@
-import { blockOrganizationAffiliation } from './workflows/block-organization-affiliation'
+import { blockProjectOrganizationAffiliations } from './workflows/block-project-organization-affiliations'
 import { cleanupDuplicateMembers } from './workflows/cleanup/duplicate-members'
 import { cleanupMembers } from './workflows/cleanup/members'
 import { cleanupOrganizations } from './workflows/cleanup/organizations'
@@ -8,6 +8,7 @@ import { findAndMergeMembersWithSameVerifiedEmailsInDifferentPlatforms } from '.
 import { fixBotMembersAffiliation } from './workflows/fix-bot-members-affiliation'
 import { fixOrgIdentitiesWithWrongUrls } from './workflows/fixOrgIdentitiesWithWrongUrls'
 import { processLLMVerifiedMerges } from './workflows/processLLMVerifiedMerges'
+import { recalculateMemberAffiliations } from './workflows/recalculate-member-affiliations'
 
 export {
   findAndMergeMembersWithSameVerifiedEmailsInDifferentPlatforms,
@@ -19,5 +20,6 @@ export {
   processLLMVerifiedMerges,
   cleanupDuplicateMembers,
   fixBotMembersAffiliation,
-  blockOrganizationAffiliation,
+  blockProjectOrganizationAffiliations,
+  recalculateMemberAffiliations,
 }

services/apps/script_executor_worker/src/workflows.ts

@@ -0,0 +1,82 @@
+import {
+  ChildWorkflowCancellationType,
+  ParentClosePolicy,
+  continueAsNew,
+  proxyActivities,
+  startChild,
+} from '@temporalio/workflow'
+
+import * as activities from '../activities'
+import { IBlockOrganizationAffiliationArgs } from '../types'
+import { chunkArray } from '../utils/common'
+
+import { recalculateMemberAffiliations } from './recalculate-member-affiliations'
+
+const {
+  fetchProjectMemberOrganizationsToBlock,
+  blockMemberOrganizationAffiliation,
+  markMemberForAffiliationRecalc,
+} = proxyActivities<typeof activities>({
+  startToCloseTimeout: '30 minutes',
+})
+
+export async function blockProjectOrganizationAffiliations(
+  args: IBlockOrganizationAffiliationArgs,
+): Promise<void> {
+  const MEMBERS_PER_RUN = 1000
+  const afterId = args.afterId ?? undefined
+
+  const memberOrganizations = await fetchProjectMemberOrganizationsToBlock(MEMBERS_PER_RUN, afterId)
+
+  if (memberOrganizations?.length === 0) {
+    console.log('No more organization members to block!')
+
+    await startChild(recalculateMemberAffiliations, {
+      workflowId: `recalculateMemberAffiliations/${Date.now()}`,
+      cancellationType: ChildWorkflowCancellationType.ABANDON,
+      parentClosePolicy: ParentClosePolicy.PARENT_CLOSE_POLICY_ABANDON,
+      retry: {
+        backoffCoefficient: 2,
+        initialInterval: 2 * 1000,
+        maximumInterval: 30 * 1000,
+      },
+      args: [
+        {
+          batchSize: 500,
+        },
+      ],
+    })
+
+    return
+  }
+
+  const CONCURRENCY = 10
+  const BATCH_SIZE = 50
+
+  // Step 1: Block affiliations in batches
+  const batches = chunkArray(memberOrganizations, BATCH_SIZE).map((chunk) =>
+    chunk.map((mo) => ({
+      memberId: mo.memberId,
+      memberOrganizationId: mo.id,
+      allowAffiliation: false,
+    })),
+  )
+
+  for (let i = 0; i < batches.length; i += CONCURRENCY) {
+    const slice = batches.slice(i, i + CONCURRENCY)
+    await Promise.all(slice.map((batch) => blockMemberOrganizationAffiliation(batch)))
+  }
+
+  // Step 2: Deduplicate memberIds and mark for affiliation recalculation
+  const uniqueMemberIds = Array.from(new Set(memberOrganizations.map((mo) => mo.memberId)))
+
+  await markMemberForAffiliationRecalc(uniqueMemberIds)
+
+  const lastProcessedId = memberOrganizations[memberOrganizations.length - 1]?.id
+
+  // Step 3: Continue pagination
+  await continueAsNew<typeof blockProjectOrganizationAffiliations>({
+    ...args,
+    afterId: lastProcessedId,
+  })
+}

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

@@ -0,0 +1,28 @@
+import { continueAsNew, proxyActivities } from '@temporalio/workflow'
+
+import * as activities from '../activities'
+import { IScriptBatchTestArgs } from '../types'
+import { chunkArray } from '../utils/common'
+
+const { getMembersForAffiliationRecalc, calculateMemberAffiliations } = proxyActivities<
+  typeof activities
+>({
+  startToCloseTimeout: '30 minutes',
+})
+
+export async function recalculateMemberAffiliations(args: IScriptBatchTestArgs): Promise<void> {
+  const MEMBERS_PER_RUN = args.batchSize ?? 200
+
+  const memberIds = await getMembersForAffiliationRecalc(MEMBERS_PER_RUN)
+
+  if (memberIds?.length === 0) {
+    console.log('No more members to recalculate affiliations!')
+    return
+  }
+
+  for (const chunk of chunkArray(memberIds, 10)) {
+    await Promise.all(chunk.map((memberId) => calculateMemberAffiliations(memberId)))
+  }
+
+  await continueAsNew<typeof recalculateMemberAffiliations>(args)
+}

services/apps/script_executor_worker/src/workflows/block-project-organization-affiliations.ts

@@ -165,18 +165,34 @@ class OrganizationRepository {
     )
   }
 
-  async findOrganizationMembers(
-    organizationId: string,
+  async fetchProjectMemberOrganizationsToBlock(
     limit: number,
-    offset: number,
-  ): Promise<IMemberOrganization[]> {
+    afterId?: string,
+  ): Promise<Pick<IMemberOrganization, 'memberId' | 'id'>[]> {
     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 },
+        select mo.id, mo."memberId"
+        from "memberOrganizations" mo
+        join organizations o on mo."organizationId" = o.id
+        where o."deletedAt" is null
+          and mo."deletedAt" is null
+          and exists (
+            select 1
+            from segments s
+            where trim(lower(o."displayName")) = trim(lower(s.name))
+          )
+          and not exists (
+            select 1
+            from "memberOrganizationAffiliationOverrides" mao
+            where mao."memberId" = mo."memberId"
+              and mao."memberOrganizationId" = mo.id
+              and mao."allowAffiliation" = false
+          )
+          ${afterId !== undefined ? 'and mo.id > $(afterId)' : ''}
+        order by mo.id
+        limit $(limit);
+      `,
+      { limit, afterId },
     )
   }
 }