feat: detect fake organizations created from email domains (CM-1411)

backend/src/api/public/v1/organizations/createOrganization.ts

@@ -31,7 +31,7 @@ export async function createOrganization(req: Request, res: Response): Promise<v
   const organizationId = await qx.tx(async (tx) => {
     const orgSource = OrganizationAttributeSource.LFX_SERVE
 
-    const organizationId = await findOrCreateOrganization(tx, orgSource, {
+    const result = await findOrCreateOrganization(tx, orgSource, {
       displayName: name,
       logo,
       identities: [
@@ -45,10 +45,12 @@ export async function createOrganization(req: Request, res: Response): Promise<v
       ],
     })
 
-    if (!organizationId) {
+    if (!result) {
       throw new InternalError('Failed to create organization')
     }
 
+    const organizationId = result.id
+
     await captureApiChange(
       req,
       organizationCreateAction(organizationId, async (captureState) => {

backend/src/database/migrations/V1788792681__fake-organization-suggestions.sql

@@ -0,0 +1,4 @@
+create table "fakeOrganizationSuggestions" (
+    "organizationId" uuid primary key not null references organizations (id) on delete cascade,
+    "createdAt" timestamp with time zone not null
+);

services/apps/data_sink_worker/src/service/activity.service.ts

@@ -20,6 +20,7 @@ import {
 } from '@crowd/common'
 import { CommonMemberService, SearchSyncWorkerEmitter } from '@crowd/common_services'
 import {
+  IFindOrCreateOrganizationResult,
   createOrUpdateRelations,
   findIdentitiesForMembers,
   findMembersByIdentities,
@@ -1162,7 +1163,7 @@ export default class ActivityService extends LoggerBase {
     // Shared org promise cache: ensures findOrCreateOrganization is called at most once per
     // unique org per batch. Concurrent member creates that reference the same org await the
     // same promise instead of firing redundant DB round trips.
-    const orgPromiseCache = new Map<string, Promise<string | undefined>>()
+    const orgPromiseCache = new Map<string, Promise<IFindOrCreateOrganizationResult | undefined>>()
 
     // find distinct members to create
     const payloadsWithoutDbMembers: IActivityProcessData[] = relevantPayloads.filter(

services/apps/data_sink_worker/src/service/member.service.ts

@@ -22,7 +22,13 @@ import {
   MEMBER_ORG_STINT_CHANGES_DATES_PREFIX,
   MEMBER_ORG_STINT_CHANGES_QUEUE,
 } from '@crowd/common_services'
-import { QueryExecutor, createMember, dbStoreQx, updateMember } from '@crowd/data-access-layer'
+import {
+  IFindOrCreateOrganizationResult,
+  QueryExecutor,
+  createMember,
+  dbStoreQx,
+  updateMember,
+} from '@crowd/data-access-layer'
 import {
   findIdentitiesForMembers,
   findMembersByIdentities,
@@ -58,6 +64,14 @@ import { IMemberCreateData, IMemberUpdateData } from './member.data'
 import MemberAttributeService from './memberAttribute.service'
 import { OrganizationService } from './organization.service'
 
+type OrgPromiseCache = Map<string, Promise<IFindOrCreateOrganizationResult | undefined>>
+
+type EmailDomainOrganization = {
+  id: string
+  source: OrganizationSource
+  created: boolean
+}
+
 /**
  * Returns a stable cache key for an org based on its verified identities, falling back to
  * displayName. Used by the org promise cache to deduplicate `findOrCreateOrganization` calls
@@ -316,7 +330,7 @@ export default class MemberService extends LoggerBase {
     integrationId: string,
     data: IMemberCreateData,
     platform: PlatformType,
-    orgPromiseCache?: Map<string, Promise<string | undefined>>,
+    orgPromiseCache?: OrgPromiseCache,
     activityTimestamp?: string,
   ): Promise<string> {
     return logExecutionTimeV2(
@@ -473,8 +487,8 @@ export default class MemberService extends LoggerBase {
             await this.startMemberBotAnalysisWithLLMWorkflow(effectiveMemberId)
           }
 
-          const organizations = []
-          const orgService = new OrganizationService(this.store, this.log)
+          const organizations: IOrganizationIdSource[] = []
+          const orgService = new OrganizationService(this.store, this.temporal, this.log)
           if (data.organizations) {
             for (const org of data.organizations) {
               // Temp fix: skip the individual-noaccount.com placeholder org to avoid
@@ -488,7 +502,7 @@ export default class MemberService extends LoggerBase {
 
               const key = orgCacheKey(org)
               const cachedOrgPromise = key ? orgPromiseCache?.get(key) : undefined
-              let orgIdPromise: Promise<string | undefined>
+              let orgIdPromise: Promise<IFindOrCreateOrganizationResult | undefined>
               if (cachedOrgPromise) {
                 orgIdPromise = cachedOrgPromise
               } else {
@@ -502,10 +516,10 @@ export default class MemberService extends LoggerBase {
                   orgIdPromise.catch(() => orgPromiseCache?.delete(key))
                 }
               }
-              const orgId = await orgIdPromise
-              if (orgId) {
+              const result = await orgIdPromise
+              if (result) {
                 organizations.push({
-                  id: orgId,
+                  id: result.id,
                   source: org.source,
                 })
               }
@@ -519,8 +533,9 @@ export default class MemberService extends LoggerBase {
           const emailIdentities = data.identities.filter(
             (i) => i.type === MemberIdentityType.EMAIL && i.verified,
           )
+          let fromEmailDomain: EmailDomainOrganization[] = []
           if (emailIdentities.length > 0) {
-            const orgs = await logExecutionTimeV2(
+            fromEmailDomain = await logExecutionTimeV2(
               () =>
                 this.assignOrganizationByEmailDomain(
                   integrationId,
@@ -533,8 +548,8 @@ export default class MemberService extends LoggerBase {
               this.log,
               'memberService -> create -> assignOrganizationByEmailDomain',
             )
-            if (orgs.length > 0) {
-              organizations.push(...orgs)
+            if (fromEmailDomain.length > 0) {
+              organizations.push(...fromEmailDomain)
             }
           }
 
@@ -562,6 +577,13 @@ export default class MemberService extends LoggerBase {
                 this.log,
                 'memberService -> create -> addToMember',
               )
+
+              const addedIds = new Set(orgsToAdd.map((org) => org.id))
+              for (const org of fromEmailDomain) {
+                if (org.created && addedIds.has(org.id)) {
+                  await orgService.startFakeOrganizationAnalysisWorkflow(org.id)
+                }
+              }
             }
           }
 
@@ -584,7 +606,7 @@ export default class MemberService extends LoggerBase {
     original: IDbMember,
     originalIdentities: IMemberIdentity[],
     platform: PlatformType,
-    orgPromiseCache?: Map<string, Promise<string | undefined>>,
+    orgPromiseCache?: OrgPromiseCache,
     activityTimestamp?: string,
   ): Promise<string | void> {
     return logExecutionTimeV2(
@@ -726,8 +748,8 @@ export default class MemberService extends LoggerBase {
             return effectiveMemberId !== id ? effectiveMemberId : undefined
           }
 
-          const organizations = []
-          const orgService = new OrganizationService(this.store, this.log)
+          const organizations: IOrganizationIdSource[] = []
+          const orgService = new OrganizationService(this.store, this.temporal, this.log)
           if (data.organizations) {
             for (const org of data.organizations) {
               // Temp fix: skip the individual-noaccount.com placeholder org to avoid
@@ -743,7 +765,7 @@ export default class MemberService extends LoggerBase {
 
               const key = orgCacheKey(org)
               const cachedOrgPromise = key ? orgPromiseCache?.get(key) : undefined
-              let orgIdPromise: Promise<string | undefined>
+              let orgIdPromise: Promise<IFindOrCreateOrganizationResult | undefined>
               if (cachedOrgPromise) {
                 orgIdPromise = cachedOrgPromise
               } else {
@@ -757,10 +779,10 @@ export default class MemberService extends LoggerBase {
                   orgIdPromise.catch(() => orgPromiseCache?.delete(key))
                 }
               }
-              const orgId = await orgIdPromise
-              if (orgId) {
+              const result = await orgIdPromise
+              if (result) {
                 organizations.push({
-                  id: orgId,
+                  id: result.id,
                   source: data.source,
                 })
               }
@@ -770,9 +792,10 @@ export default class MemberService extends LoggerBase {
           const emailIdentities = data.identities.filter(
             (i) => i.verified && i.type === MemberIdentityType.EMAIL,
           )
+          let fromEmailDomain: EmailDomainOrganization[] = []
           if (emailIdentities.length > 0) {
             this.log.trace({ memberId: id }, 'Assigning organization by email domain!')
-            const orgs = await logExecutionTimeV2(
+            fromEmailDomain = await logExecutionTimeV2(
               () =>
                 this.assignOrganizationByEmailDomain(
                   integrationId,
@@ -785,8 +808,8 @@ export default class MemberService extends LoggerBase {
               this.log,
               'memberService -> update -> assignOrganizationByEmailDomain',
             )
-            if (orgs.length > 0) {
-              organizations.push(...orgs)
+            if (fromEmailDomain.length > 0) {
+              organizations.push(...fromEmailDomain)
             }
           }
 
@@ -816,6 +839,13 @@ export default class MemberService extends LoggerBase {
                 this.log,
                 'memberService -> update -> addToMember',
               )
+
+              const addedIds = new Set(orgsToAdd.map((org) => org.id))
+              for (const org of fromEmailDomain) {
+                if (org.created && addedIds.has(org.id)) {
+                  await orgService.startFakeOrganizationAnalysisWorkflow(org.id)
+                }
+              }
             }
           }
 
@@ -833,13 +863,13 @@ export default class MemberService extends LoggerBase {
   public async assignOrganizationByEmailDomain(
     integrationId: string,
     emails: string[],
-    orgPromiseCache?: Map<string, Promise<string | undefined>>,
+    orgPromiseCache?: OrgPromiseCache,
     memberId?: string,
     activityTimestamp?: string,
     isBotMember = false,
-  ): Promise<IOrganizationIdSource[]> {
-    const orgService = new OrganizationService(this.store, this.log)
-    const organizations: IOrganizationIdSource[] = []
+  ): Promise<EmailDomainOrganization[]> {
+    const orgService = new OrganizationService(this.store, this.temporal, this.log)
+    const organizations: EmailDomainOrganization[] = []
     const emailDomains = new Set<string>()
 
     // Collect unique domains
@@ -876,7 +906,7 @@ export default class MemberService extends LoggerBase {
       }
       const key = orgCacheKey(org)
       const cachedOrgPromise = key ? orgPromiseCache?.get(key) : undefined
-      let orgIdPromise: Promise<string | undefined>
+      let orgIdPromise: Promise<IFindOrCreateOrganizationResult | undefined>
       if (cachedOrgPromise) {
         orgIdPromise = cachedOrgPromise
       } else {
@@ -890,15 +920,16 @@ export default class MemberService extends LoggerBase {
           orgIdPromise.catch(() => orgPromiseCache?.delete(key))
         }
       }
-      const orgId = await orgIdPromise
-      if (orgId) {
+      const result = await orgIdPromise
+      if (result) {
         organizations.push({
-          id: orgId,
+          id: result.id,
           source: orgSource,
+          created: result.created,
         })
 
         if (memberId && activityTimestamp && !isBotMember) {
-          await this.bufferMemberOrganizationActivityDates(memberId, orgId, activityTimestamp)
+          await this.bufferMemberOrganizationActivityDates(memberId, result.id, activityTimestamp)
         }
       }
     }

services/apps/data_sink_worker/src/service/organization.service.ts

@@ -1,22 +1,31 @@
+import { DEFAULT_TENANT_ID } from '@crowd/common'
 import {
   changeMemberOrganizationAffiliationOverrides,
   fetchManyOrganizationAffiliationPolicies,
 } from '@crowd/data-access-layer'
 import { DbStore } from '@crowd/data-access-layer/src/database'
 import { deleteMemberSegmentAffiliations } from '@crowd/data-access-layer/src/member_segment_affiliations'
 import {
+  IFindOrCreateOrganizationResult,
   addOrgsToMember,
   addOrgsToSegments,
   findMemberOrganizations,
   findOrCreateOrganization,
 } from '@crowd/data-access-layer/src/organizations'
 import { dbStoreQx } from '@crowd/data-access-layer/src/queryExecutor'
 import { Logger, LoggerBase } from '@crowd/logging'
-import { IMemberOrganization, IOrganization, IOrganizationIdSource } from '@crowd/types'
+import { Client as TemporalClient } from '@crowd/temporal'
+import {
+  IMemberOrganization,
+  IOrganization,
+  IOrganizationIdSource,
+  TemporalWorkflowId,
+} from '@crowd/types'
 
 export class OrganizationService extends LoggerBase {
   constructor(
     private readonly store: DbStore,
+    private readonly temporal: TemporalClient,
     parentLog: Logger,
   ) {
     super(parentLog)
@@ -26,7 +35,7 @@ export class OrganizationService extends LoggerBase {
     source: string,
     integrationId: string,
     data: IOrganization,
-  ): Promise<string | undefined> {
+  ): Promise<IFindOrCreateOrganizationResult | undefined> {
     return this.store.transactionally(async (txStore) => {
       const qe = dbStoreQx(txStore)
       return findOrCreateOrganization(qe, source, data, integrationId, true)
@@ -82,4 +91,30 @@ export class OrganizationService extends LoggerBase {
 
     return findMemberOrganizations(qe, memberId, organizationId)
   }
+
+  public async startFakeOrganizationAnalysisWorkflow(organizationId: string): Promise<void> {
+    try {
+      await this.temporal.workflow.start('fakeOrganizationAnalysisWithLLM', {
+        taskQueue: 'profiles',
+        workflowId: `${TemporalWorkflowId.FAKE_ORGANIZATION_ANALYSIS_WITH_LLM}/${organizationId}`,
+        retry: {
+          maximumAttempts: 10,
+        },
+        args: [{ organizationId }],
+        searchAttributes: {
+          TenantId: [DEFAULT_TENANT_ID],
+        },
+      })
+    } catch (err) {
+      if (err.name === 'WorkflowExecutionAlreadyStartedError') {
+        this.log.info(
+          { organizationId },
+          'Fake organization analysis workflow already started, skipping',
+        )
+        return
+      }
+
+      throw err
+    }
+  }
 }

services/apps/members_enrichment_worker/src/activities/enrichment.ts

@@ -472,7 +472,7 @@ export async function updateMemberUsingSquashedPayload(
         try {
           // Keep the org write in a savepoint: if this identity is already verified
           // on another org, we can recover without aborting the member update transaction.
-          orgId = await qx.tx((trnx) => findOrCreateOrganization(trnx, orgSource, orgPayload))
+          orgId = (await qx.tx((trnx) => findOrCreateOrganization(trnx, orgSource, orgPayload)))?.id
         } catch (error) {
           const constraint = 'uix_organizationIdentities_plat_val_typ_tenantId_verified'
           const dbError = error as { constraint?: string; detail?: string }
@@ -535,12 +535,14 @@ export async function updateMemberUsingSquashedPayload(
               ),
           )
 
-          orgId = await qx.tx((trnx) =>
-            findOrCreateOrganization(trnx, orgSource, {
-              ...orgPayload,
-              identities: retryIdentities,
-            }),
-          )
+          orgId = (
+            await qx.tx((trnx) =>
+              findOrCreateOrganization(trnx, orgSource, {
+                ...orgPayload,
+                identities: retryIdentities,
+              }),
+            )
+          )?.id
 
           if (orgId) {
             const mergeSuggestionsRepo = new OrganizationMergeSuggestionsRepository(