fix: data sink worker optimizations (CM-1054) (#3952)

Signed-off-by: Uroš Marolt <uros@marolt.me>

Author: themarolt

backend/src/database/migrations/U1774430554__covering-indexes-for-member-identity-lookups.sql

@@ -0,0 +1,36 @@
+-- Query: findMembersByVerifiedUsernames
+--
+-- Joins memberIdentities on (platform, lower(value)) with WHERE:
+--   verified = true AND type = 'username' AND deletedAt IS NULL
+--
+-- The existing idx_memberIdentities_platform_type_lower_value_memberId is
+-- missing verified = true in its partial condition, so PostgreSQL must
+-- heap-fetch every row to recheck verified, which is expensive when there
+-- are many unverified identities.
+--
+-- This index adds verified = true to the partial condition and includes memberId
+-- so the join to members can read memberId from the index without a heap fetch.
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "idx_memberIdentities_verified_username_platform_lower_value"
+    ON "memberIdentities" (platform, lower(value), "memberId")
+    WHERE verified = true
+      AND type = 'username'
+      AND "deletedAt" IS NULL;
+
+-- Query: findMembersByVerifiedEmails
+--
+-- Joins memberIdentities on lower(value) with WHERE:
+--   verified = true AND type = 'email' AND deletedAt IS NULL
+--
+-- The existing idx_memberIdentities_verified_email_lower_value has the right
+-- partial condition but only stores lower(value) — no memberId. Every matched
+-- index entry requires a heap fetch to get memberId for the join to members.
+-- Under concurrent insert/update load, those heap fetches queue behind buffer
+-- pin locks, causing multi-second delays even for small inputs.
+--
+-- This index adds memberId so the join to members can proceed without
+-- touching the memberIdentities heap pages.
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "idx_memberIdentities_verified_email_lower_value_memberid"
+    ON "memberIdentities" (lower(value), "memberId")
+    WHERE verified = true
+      AND type = 'email'
+      AND "deletedAt" IS NULL;

backend/src/database/migrations/V1774430554__covering-indexes-for-member-identity-lookups.sql

@@ -1050,6 +1050,10 @@ export default class ActivityService extends LoggerBase {
     const preparedActivities: IActivityPrepareForUpsertResult[] = []
 
     const memberService = new MemberService(this.pgStore, this.redisClient, this.temporal, this.log)
+    // 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>>()
 
     // find distinct members to create
     const payloadsWithoutDbMembers: IActivityProcessData[] = relevantPayloads.filter(
@@ -1145,6 +1149,8 @@ export default class ActivityService extends LoggerBase {
               reach: value.member.reach,
             },
             value.platform,
+            undefined,
+            orgPromiseCache,
           )
           .then((memberId) => {
             // map ids for members
@@ -1288,6 +1294,8 @@ export default class ActivityService extends LoggerBase {
                 payload.dbMember,
                 dbMemberIdentities.get(payload.dbMember.id),
                 payload.platform,
+                undefined,
+                orgPromiseCache,
               )
               .then(() => {
                 payload.memberId = payload.dbMember.id
@@ -1344,6 +1352,8 @@ export default class ActivityService extends LoggerBase {
                 payload.dbObjectMember,
                 dbMemberIdentities.get(payload.dbObjectMember.id),
                 payload.platform,
+                undefined,
+                orgPromiseCache,
               )
               .then(() => {
                 payload.objectMemberId = payload.dbObjectMember.id
@@ -1538,7 +1548,7 @@ export default class ActivityService extends LoggerBase {
       this.log.trace(
         `[ACTIVITY] Upserting ${relationsToUpsert.length} activity relations (filtered from ${preparedForUpsert.length}, skipped ${skippedCount})`,
       )
-      await createOrUpdateRelations(this.pgQx, relationsToUpsert)
+      await createOrUpdateRelations(this.pgQx, relationsToUpsert, true)
     } else {
       this.log.trace(
         `[ACTIVITY] No activity relations need updating (all ${preparedForUpsert.length} would only update updatedAt)`,
@@ -1580,28 +1590,36 @@ export default class ActivityService extends LoggerBase {
           `${prepared.payload.platform}:${prepared.payload.channel}:${prepared.payload.segmentId}`,
         )
       }
+    }
 
-      await this.searchSyncWorkerEmitter.triggerMemberSync(
-        prepared.payload.memberId,
-        onboarding,
-        prepared.payload.segmentId,
-      )
-
-      if (prepared.payload.objectMemberId) {
-        await this.searchSyncWorkerEmitter.triggerMemberSync(
-          prepared.payload.objectMemberId,
-          onboarding,
-          prepared.payload.segmentId,
-        )
-      }
+    const orgIds = preparedForUpsert
+      .map((p) => p.payload.organizationId)
+      .filter((id): id is string => !!id)
+    if (orgIds.length > 0) {
+      await this.redisClient.sAdd('organizationIdsForAggComputation', orgIds)
+    }
 
-      if (prepared.payload.organizationId) {
-        await this.redisClient.sAdd(
-          'organizationIdsForAggComputation',
-          prepared.payload.organizationId,
-        )
+    // Deduplicate member sync triggers — a member may appear in many activities in the
+    // same batch. Emit once per unique (memberId, segmentId) pair.
+    const memberSyncKeys = new Set<string>()
+    const memberSyncPromises: Promise<void>[] = []
+    for (const prepared of preparedForUpsert) {
+      for (const memberId of [prepared.payload.memberId, prepared.payload.objectMemberId]) {
+        if (!memberId) continue
+        const key = `${memberId}:${prepared.payload.segmentId}`
+        if (!memberSyncKeys.has(key)) {
+          memberSyncKeys.add(key)
+          memberSyncPromises.push(
+            this.searchSyncWorkerEmitter.triggerMemberSync(
+              memberId,
+              onboarding,
+              prepared.payload.segmentId,
+            ),
+          )
+        }
       }
     }
+    await Promise.all(memberSyncPromises)
 
     for (const prepared of preparedActivities) {
       resultMap.set(prepared.resultId, { success: true })

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

@@ -28,6 +28,7 @@ import { Client as TemporalClient } from '@crowd/temporal'
 import {
   IMemberData,
   IMemberIdentity,
+  IOrganization,
   IOrganizationIdSource,
   MemberAttributeName,
   MemberBotDetection,
@@ -43,6 +44,22 @@ import { IMemberCreateData, IMemberUpdateData } from './member.data'
 import MemberAttributeService from './memberAttribute.service'
 import { OrganizationService } from './organization.service'
 
+/**
+ * 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
+ * across members in the same batch.
+ */
+function orgCacheKey(org: IOrganization): string | null {
+  const verified = (org.identities ?? [])
+    .filter((i) => i.verified)
+    .map((i) => `${i.platform}:${i.type}:${i.value.toLowerCase()}`)
+    .sort()
+    .join('|')
+  if (verified) return verified
+  if (org.displayName) return `name:${org.displayName.toLowerCase()}`
+  return null
+}
+
 export default class MemberService extends LoggerBase {
   private readonly memberRepo: MemberRepository
   private readonly pgQx: QueryExecutor
@@ -67,6 +84,7 @@ export default class MemberService extends LoggerBase {
     data: IMemberCreateData,
     platform: PlatformType,
     releaseMemberLock?: () => Promise<void>,
+    orgPromiseCache?: Map<string, Promise<string | undefined>>,
   ): Promise<string> {
     return logExecutionTimeV2(
       async () => {
@@ -188,11 +206,31 @@ export default class MemberService extends LoggerBase {
           const orgService = new OrganizationService(this.store, this.log)
           if (data.organizations) {
             for (const org of data.organizations) {
-              const id = await logExecutionTimeV2(
-                () => orgService.findOrCreate(platform, integrationId, org),
-                this.log,
-                'memberService -> create -> findOrCreateOrg',
-              )
+              // Temp fix: skip the individual-noaccount.com placeholder org to avoid
+              // hot-row contention on the organizations table. Permanent fix is in
+              // tncTransformerBase.ts to stop emitting this org entirely.
+              if (
+                org.identities?.some((i) => i.verified && i.value === 'individual-noaccount.com')
+              ) {
+                continue
+              }
+
+              const key = orgCacheKey(org)
+              let orgIdPromise: Promise<string | undefined>
+              if (key && orgPromiseCache?.has(key)) {
+                orgIdPromise = orgPromiseCache.get(key)
+              } else {
+                orgIdPromise = logExecutionTimeV2(
+                  () => orgService.findOrCreate(platform, integrationId, org),
+                  this.log,
+                  'memberService -> create -> findOrCreateOrg',
+                )
+                if (key) {
+                  orgPromiseCache?.set(key, orgIdPromise)
+                  orgIdPromise.catch(() => orgPromiseCache?.delete(key))
+                }
+              }
+              const id = await orgIdPromise
               organizations.push({
                 id,
                 source: org.source,
@@ -209,6 +247,7 @@ export default class MemberService extends LoggerBase {
                 this.assignOrganizationByEmailDomain(
                   integrationId,
                   emailIdentities.map((i) => i.value),
+                  orgPromiseCache,
                 ),
               this.log,
               'memberService -> create -> assignOrganizationByEmailDomain',
@@ -220,7 +259,6 @@ export default class MemberService extends LoggerBase {
 
           if (organizations.length > 0) {
             const uniqOrgs = uniqby(organizations, 'id')
-            const orgService = new OrganizationService(this.store, this.log)
 
             const orgsToAdd = (
               await Promise.all(
@@ -266,6 +304,7 @@ export default class MemberService extends LoggerBase {
     originalIdentities: IMemberIdentity[],
     platform: PlatformType,
     releaseMemberLock?: () => Promise<void>,
+    orgPromiseCache?: Map<string, Promise<string | undefined>>,
   ): Promise<void> {
     await logExecutionTimeV2(
       async () => {
@@ -398,13 +437,33 @@ export default class MemberService extends LoggerBase {
           const orgService = new OrganizationService(this.store, this.log)
           if (data.organizations) {
             for (const org of data.organizations) {
+              // Temp fix: skip the individual-noaccount.com placeholder org to avoid
+              // hot-row contention on the organizations table. Permanent fix is in
+              // tncTransformerBase.ts to stop emitting this org entirely.
+              if (
+                org.identities?.some((i) => i.verified && i.value === 'individual-noaccount.com')
+              ) {
+                continue
+              }
+
               this.log.trace({ memberId: id }, 'Finding or creating organization!')
 
-              const orgId = await logExecutionTimeV2(
-                () => orgService.findOrCreate(platform, integrationId, org),
-                this.log,
-                'memberService -> update -> findOrCreateOrg',
-              )
+              const key = orgCacheKey(org)
+              let orgIdPromise: Promise<string | undefined>
+              if (key && orgPromiseCache?.has(key)) {
+                orgIdPromise = orgPromiseCache.get(key)
+              } else {
+                orgIdPromise = logExecutionTimeV2(
+                  () => orgService.findOrCreate(platform, integrationId, org),
+                  this.log,
+                  'memberService -> update -> findOrCreateOrg',
+                )
+                if (key) {
+                  orgPromiseCache?.set(key, orgIdPromise)
+                  orgIdPromise.catch(() => orgPromiseCache?.delete(key))
+                }
+              }
+              const orgId = await orgIdPromise
               organizations.push({
                 id: orgId,
                 source: data.source,
@@ -422,6 +481,7 @@ export default class MemberService extends LoggerBase {
                 this.assignOrganizationByEmailDomain(
                   integrationId,
                   emailIdentities.map((i) => i.value),
+                  orgPromiseCache,
                 ),
               this.log,
               'memberService -> update -> assignOrganizationByEmailDomain',
@@ -433,7 +493,6 @@ export default class MemberService extends LoggerBase {
 
           if (organizations.length > 0) {
             const uniqOrgs = uniqby(organizations, 'id')
-            const orgService = new OrganizationService(this.store, this.log)
 
             this.log.trace({ memberId: id }, 'Finding member organizations!')
             const orgsToAdd = (
@@ -473,6 +532,7 @@ export default class MemberService extends LoggerBase {
   public async assignOrganizationByEmailDomain(
     integrationId: string,
     emails: string[],
+    orgPromiseCache?: Map<string, Promise<string | undefined>>,
   ): Promise<IOrganizationIdSource[]> {
     const orgService = new OrganizationService(this.store, this.log)
     const organizations: IOrganizationIdSource[] = []
@@ -494,26 +554,38 @@ export default class MemberService extends LoggerBase {
     // Assign member to organization based on email domain
     for (const domain of emailDomains) {
       const orgSource = OrganizationSource.EMAIL_DOMAIN
-      const orgId = await orgService.findOrCreate(
-        OrganizationAttributeSource.EMAIL,
-        integrationId,
-        {
-          attributes: {
-            name: {
-              integration: [domain],
-            },
+      const org: IOrganization = {
+        attributes: {
+          name: {
+            integration: [domain],
           },
-          identities: [
-            {
-              value: domain,
-              type: OrganizationIdentityType.PRIMARY_DOMAIN,
-              platform: 'email',
-              verified: true,
-              source: orgSource,
-            },
-          ],
         },
-      )
+        identities: [
+          {
+            value: domain,
+            type: OrganizationIdentityType.PRIMARY_DOMAIN,
+            platform: 'email',
+            verified: true,
+            source: orgSource,
+          },
+        ],
+      }
+      const key = orgCacheKey(org)
+      let orgIdPromise: Promise<string | undefined>
+      if (key && orgPromiseCache?.has(key)) {
+        orgIdPromise = orgPromiseCache.get(key)
+      } else {
+        orgIdPromise = orgService.findOrCreate(
+          OrganizationAttributeSource.EMAIL,
+          integrationId,
+          org,
+        )
+        if (key) {
+          orgPromiseCache?.set(key, orgIdPromise)
+          orgIdPromise.catch(() => orgPromiseCache?.delete(key))
+        }
+      }
+      const orgId = await orgIdPromise
       if (orgId) {
         organizations.push({
           id: orgId,