fix: refresh segment-only members when blocking orgs

backend/src/services/organizationService.ts

@@ -19,15 +19,15 @@ import {
   mergeObjects,
   normalizeHostname,
 } from '@crowd/common'
-import { unmergeRoles } from '@crowd/common_services'
+import { signalMemberUpdate, unmergeRoles } from '@crowd/common_services'
 import {
   addMemberRole,
   moveMembersBetweenOrganizations,
   removeMemberRole,
 } from '@crowd/data-access-layer'
 import { hasLfxMembership } from '@crowd/data-access-layer/src/lfx_memberships'
 import { applyOrganizationAffiliationPolicyToMembers } from '@crowd/data-access-layer/src/member-organization-affiliation'
-import { deleteMemberSegmentAffiliations } from '@crowd/data-access-layer/src/member_segment_affiliations'
+import { deleteOrganizationSegmentAffiliations } from '@crowd/data-access-layer/src/member_segment_affiliations'
 import {
   addMergeAction,
   queryMergeActions,
@@ -1093,6 +1093,7 @@ export default class OrganizationService extends LoggerBase {
   ) {
     let tx
     let recalculateAffiliations = false
+    let segmentOnlyMemberIds: string[] = []
 
     try {
       const repoOptions = await SequelizeRepository.createTransactionalRepositoryOptions(
@@ -1185,7 +1186,7 @@ export default class OrganizationService extends LoggerBase {
       ) {
         await applyOrganizationAffiliationPolicyToMembers(qx, record.id, !data.isAffiliationBlocked)
         if (data.isAffiliationBlocked) {
-          await deleteMemberSegmentAffiliations(qx, { organizationId: record.id })
+          segmentOnlyMemberIds = await deleteOrganizationSegmentAffiliations(qx, record.id)
         }
         recalculateAffiliations = true
       }
@@ -1203,6 +1204,10 @@ export default class OrganizationService extends LoggerBase {
         })
       }
 
+      for (const memberId of segmentOnlyMemberIds) {
+        await signalMemberUpdate(this.options.temporal, memberId)
+      }
+
       return record
     } catch (error) {
       if (tx) {

backend/src/services/segmentService.ts

@@ -1,14 +1,15 @@
 import { Transaction } from 'sequelize'
 
 import { Error400, generateOrganizationNameVariants, validateNonLfSlug } from '@crowd/common'
+import { signalMemberUpdate } from '@crowd/common_services'
 import {
   QueryExecutor,
   findManyOrganizationsByNames,
   updateOrganization,
 } from '@crowd/data-access-layer'
 import { ICreateInsightsProject, findBySlug } from '@crowd/data-access-layer/src/collections'
 import { applyOrganizationAffiliationPolicyToMembers } from '@crowd/data-access-layer/src/member-organization-affiliation'
-import { deleteMemberSegmentAffiliations } from '@crowd/data-access-layer/src/member_segment_affiliations'
+import { deleteOrganizationSegmentAffiliations } from '@crowd/data-access-layer/src/member_segment_affiliations'
 import {
   buildSegmentActivityTypes,
   getSegmentSubprojects,
@@ -37,6 +38,17 @@ import OrganizationService from './organizationService'
 interface UnnestedActivityTypes {
   [key: string]: any
 }
+
+interface BlockedOrganizationAffiliations {
+  organizationIds: string[]
+  segmentOnlyMemberIds: string[]
+}
+
+const NO_BLOCKED_AFFILIATIONS: BlockedOrganizationAffiliations = {
+  organizationIds: [],
+  segmentOnlyMemberIds: [],
+}
+
 export default class SegmentService extends LoggerBase {
   options: IServiceOptions
 
@@ -150,26 +162,16 @@ export default class SegmentService extends LoggerBase {
 
       // Only apply project-org affiliation blocking for LF segments.
       // Use the persisted segment flag (not raw input) as the source of truth.
-      const orgIds = projectGroup.isLF
+      const blockedAffiliations = projectGroup.isLF
         ? await this.blockOrganizationAffiliationIfSegmentNameMatches(
             projectGroup.name,
             transaction,
           )
-        : []
+        : NO_BLOCKED_AFFILIATIONS
 
       await SequelizeRepository.commitTransaction(transaction)
 
-      if (orgIds.length > 0) {
-        const organizationService = new OrganizationService(this.options)
-
-        for (const orgId of orgIds) {
-          // Trigger org update workflow to recalculate affiliations
-          await organizationService.startOrganizationUpdateWorkflow(orgId, {
-            syncToOpensearch: true,
-            recalculateAffiliations: true,
-          })
-        }
-      }
+      await this.recalculateBlockedAffiliations(blockedAffiliations)
 
       return await this.findById(projectGroup.id)
     } catch (error) {
@@ -232,23 +234,13 @@ export default class SegmentService extends LoggerBase {
 
       // Only apply project-org affiliation blocking for LF segments.
       // Use the persisted segment flag (not raw input) as the source of truth.
-      const orgIds = project.isLF
+      const blockedAffiliations = project.isLF
         ? await this.blockOrganizationAffiliationIfSegmentNameMatches(project.name, transaction)
-        : []
+        : NO_BLOCKED_AFFILIATIONS
 
       await SequelizeRepository.commitTransaction(transaction)
 
-      if (orgIds.length > 0) {
-        const organizationService = new OrganizationService(this.options)
-
-        for (const orgId of orgIds) {
-          // Trigger org update workflow to recalculate affiliations
-          await organizationService.startOrganizationUpdateWorkflow(orgId, {
-            syncToOpensearch: true,
-            recalculateAffiliations: true,
-          })
-        }
-      }
+      await this.recalculateBlockedAffiliations(blockedAffiliations)
 
       return await this.findById(project.id)
     } catch (error) {
@@ -264,22 +256,13 @@ export default class SegmentService extends LoggerBase {
     try {
       const subproject = await this.createSubprojectInternal(data, qx, transaction)
 
-      const orgIds = subproject.isLF
+      const blockedAffiliations = subproject.isLF
         ? await this.blockOrganizationAffiliationIfSegmentNameMatches(subproject.name, transaction)
-        : []
+        : NO_BLOCKED_AFFILIATIONS
 
       await SequelizeRepository.commitTransaction(transaction)
 
-      if (orgIds.length > 0) {
-        const organizationService = new OrganizationService(this.options)
-
-        for (const orgId of orgIds) {
-          await organizationService.startOrganizationUpdateWorkflow(orgId, {
-            syncToOpensearch: true,
-            recalculateAffiliations: true,
-          })
-        }
-      }
+      await this.recalculateBlockedAffiliations(blockedAffiliations)
 
       return await this.findById(subproject.id)
     } catch (error) {
@@ -719,7 +702,7 @@ export default class SegmentService extends LoggerBase {
   private async blockOrganizationAffiliationIfSegmentNameMatches(
     segmentName: string,
     transaction: Transaction,
-  ): Promise<string[]> {
+  ): Promise<BlockedOrganizationAffiliations> {
     const qx = SequelizeRepository.getQueryExecutor({
       ...this.options,
       transaction,
@@ -731,26 +714,44 @@ export default class SegmentService extends LoggerBase {
       generateOrganizationNameVariants(segmentName),
     )
 
-    if (organizations.length === 0) {
-      return []
+    const result: BlockedOrganizationAffiliations = {
+      organizationIds: [],
+      segmentOnlyMemberIds: [],
     }
 
-    const result: string[] = []
-
     for (const o of organizations) {
       if (!o.isAffiliationBlocked) {
         const updatedOrgId = await updateOrganization(qx, o.id, { isAffiliationBlocked: true })
         if (updatedOrgId) {
           await applyOrganizationAffiliationPolicyToMembers(qx, updatedOrgId, false)
-          await deleteMemberSegmentAffiliations(qx, { organizationId: updatedOrgId })
-          result.push(updatedOrgId)
+          const memberIds = await deleteOrganizationSegmentAffiliations(qx, updatedOrgId)
+          result.organizationIds.push(updatedOrgId)
+          result.segmentOnlyMemberIds.push(...memberIds)
         }
       }
     }
 
     return result
   }
 
+  private async recalculateBlockedAffiliations({
+    organizationIds,
+    segmentOnlyMemberIds,
+  }: BlockedOrganizationAffiliations): Promise<void> {
+    const organizationService = new OrganizationService(this.options)
+
+    for (const orgId of organizationIds) {
+      await organizationService.startOrganizationUpdateWorkflow(orgId, {
+        syncToOpensearch: true,
+        recalculateAffiliations: true,
+      })
+    }
+
+    for (const memberId of new Set(segmentOnlyMemberIds)) {
+      await signalMemberUpdate(this.options.temporal, memberId)
+    }
+  }
+
   /**
    * Throws an appropriate error message when a segment conflict is detected.
    * This method dynamically generates error messages based on the existing conflicting segment,

services/apps/profiles_worker/src/activities/organization/fakeAnalysis.ts

@@ -17,7 +17,7 @@ import {
   updateOrganization,
 } from '@crowd/data-access-layer'
 import { applyOrganizationAffiliationPolicyToMembers } from '@crowd/data-access-layer/src/member-organization-affiliation'
-import { deleteMemberSegmentAffiliations } from '@crowd/data-access-layer/src/member_segment_affiliations'
+import { deleteOrganizationSegmentAffiliations } from '@crowd/data-access-layer/src/member_segment_affiliations'
 import {
   IAttributes,
   IMemberIdentity,
@@ -29,6 +29,7 @@ import {
 } from '@crowd/types'
 
 import { svc } from '../../main'
+import { triggerMemberAffiliationsRefresh } from '../member/memberUpdate'
 
 export async function getOrganizationForFakeAnalysis(
   organizationId: string,
@@ -131,7 +132,11 @@ export async function markOrganizationAsFake(organizationId: string): Promise<vo
 
   await updateOrganization(qx, organizationId, { isAffiliationBlocked: true })
   await applyOrganizationAffiliationPolicyToMembers(qx, organizationId, false)
-  await deleteMemberSegmentAffiliations(qx, { organizationId })
+  const segmentOnlyMemberIds = await deleteOrganizationSegmentAffiliations(qx, organizationId)
+
+  for (const memberId of segmentOnlyMemberIds) {
+    await triggerMemberAffiliationsRefresh(memberId)
+  }
 
   const workflowId = `${TemporalWorkflowId.ORGANIZATION_UPDATE}/${organizationId}`
 

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

@@ -1,6 +1,6 @@
 import { OrganizationField, findOrgById, pgpQx, updateOrganization } from '@crowd/data-access-layer'
 import { changeMemberOrganizationAffiliationOverrides } from '@crowd/data-access-layer/src/member-organization-affiliation'
-import { deleteMemberSegmentAffiliations } from '@crowd/data-access-layer/src/member_segment_affiliations'
+import { deleteOrganizationSegmentAffiliations } from '@crowd/data-access-layer/src/member_segment_affiliations'
 import OrganizationRepository from '@crowd/data-access-layer/src/old/apps/script_executor_worker/organization.repo'
 import { IChangeAffiliationOverrideData, IMemberOrganization } from '@crowd/types'
 
@@ -36,7 +36,12 @@ export async function setOrganizationAffiliationPolicyIfNotBlocked(
     }
 
     await updateOrganization(qx, organizationId, { isAffiliationBlocked: true })
-    await deleteMemberSegmentAffiliations(qx, { organizationId })
+    const segmentOnlyMemberIds = await deleteOrganizationSegmentAffiliations(qx, organizationId)
+
+    if (segmentOnlyMemberIds.length > 0) {
+      await markMemberForAffiliationRecalc(segmentOnlyMemberIds)
+    }
+
     return false
   } catch (error) {
     svc.log.error(error, 'Error setting organization affiliation policy!')

services/libs/data-access-layer/src/member_segment_affiliations/index.ts

@@ -66,6 +66,37 @@ export async function deleteMemberSegmentAffiliations(
   return qx.result(query, params)
 }
 
+// Returns only members with no active memberOrganizations row for the org; those with one are
+// already refreshed by the organization's affiliation recalculation.
+export async function deleteOrganizationSegmentAffiliations(
+  qx: QueryExecutor,
+  organizationId: string,
+): Promise<string[]> {
+  const rows: { memberId: string }[] = await qx.select(
+    `
+      WITH deleted AS (
+        UPDATE "memberSegmentAffiliations"
+        SET "deletedAt" = NOW()
+        WHERE "organizationId" = $(organizationId)
+          AND "deletedAt" IS NULL
+        RETURNING "memberId"
+      )
+      SELECT DISTINCT d."memberId"
+      FROM deleted d
+      WHERE NOT EXISTS (
+        SELECT 1
+        FROM "memberOrganizations" mo
+        WHERE mo."memberId" = d."memberId"
+          AND mo."organizationId" = $(organizationId)
+          AND mo."deletedAt" IS NULL
+      )
+    `,
+    { organizationId },
+  )
+
+  return rows.map((row) => row.memberId)
+}
+
 export async function findMemberAffiliations(
   qx: QueryExecutor,
   memberId: string,