Merge branch 'main' into chore/claude-code-setup

Author: skwowet

backend/src/database/flyway_migrate.sh

@@ -10,6 +10,7 @@ flyway \
     -password="$PGPASSWORD" \
     -connectRetries=60 \
     -outOfOrder=true \
+    -mixed=true \
     -placeholderReplacement=false \
     -schemas=public \
     -X \

backend/src/database/migrations/U1774609007__data-sink-worker-optimizations.sql

@@ -0,0 +1,11 @@
+-- Drop 4 unused activityRelations indexes (already dropped on prod 2026-03-27,
+-- see ACTIVITYRELATIONS_INDEX_CLEANUP.md — IF EXISTS guards for idempotency)
+alter table "activityRelations" drop constraint if exists "activityRelations_activityId_memberId_key";
+
+drop index concurrently if exists "ix_activityRelations_memberId_segmentId_include";
+drop index concurrently if exists "ix_activityRelations_organizationId_segmentId_include";
+drop index concurrently if exists "ix_activityRelations_platform_username";
+
+create index concurrently if not exists idx_osa_org_segment_membercount
+    on "organizationSegmentsAgg" ("organizationId", "segmentId")
+    include ("memberCount");

backend/src/database/migrations/V1774609007__data-sink-worker-optimizations.sql

@@ -10,6 +10,7 @@ flyway \
     -password="$PGPASSWORD" \
     -connectRetries=60 \
     -outOfOrder=true \
+    -mixed=true \
     -placeholderReplacement=false \
     -schemas=public \
     -X \

backend/src/product/flyway_migrate.sh

@@ -28,7 +28,6 @@ import {
 } from '@crowd/data-access-layer'
 import { IDbActivityRelation } from '@crowd/data-access-layer/src/activityRelations/types'
 import { DbStore, arePrimitivesDbEqual } from '@crowd/data-access-layer/src/database'
-import { getMemberNoMerge } from '@crowd/data-access-layer/src/member_merge'
 import {
   IActivityRelationCreateOrUpdateData,
   IDbActivity,
@@ -56,7 +55,7 @@ import {
 } from '@crowd/types'
 
 import { IActivityUpdateData, ISentimentActivityInput } from './activity.data'
-import MemberService from './member.service'
+import MemberService, { mergeIfAllowed } from './member.service'
 import { IProcessActivityResult } from './types'
 
 /* eslint-disable @typescript-eslint/no-explicit-any */
@@ -293,6 +292,28 @@ export default class ActivityService extends LoggerBase {
         }
       }
 
+      // When activity.username is set but differs from the member's platform identity value,
+      // override it so the member lookup and the identity insert use the same key.
+      // Example: git activities set activity.username to the author display name (e.g. "John Doe")
+      // while the identity stores the email (e.g. "john.doe@example.com"). Without this correction
+      // the lookup misses the existing member, creating an unnecessary orphan member.
+      if (username && member) {
+        const platformIdentity = member.identities.find(
+          (i) =>
+            i.platform === platform &&
+            i.type === MemberIdentityType.USERNAME &&
+            i.value &&
+            i.verified,
+        )
+        if (platformIdentity && platformIdentity.value !== username) {
+          this.log.debug(
+            { platform, originalUsername: username, correctedUsername: platformIdentity.value },
+            'Overriding activity.username with member platform identity value',
+          )
+          activity.username = platformIdentity.value
+        }
+      }
+
       member.identities = member.identities.filter((i) => i.value)
 
       if (!username) {
@@ -1721,32 +1742,24 @@ export default class ActivityService extends LoggerBase {
         const originalId = metadata.memberWithIdentity as string
         const targetId = metadata.memberIdToUpdate as string
 
-        // but first check memberNoMerge table
-        const noMergeMemberIds = await getMemberNoMerge(this.pgQx, [originalId, targetId])
-
-        const noMerge = singleOrDefault(
-          noMergeMemberIds,
-          (m) =>
-            (m.memberId === originalId && m.noMergeId === targetId) ||
-            (m.memberId === targetId && m.noMergeId === originalId),
-        )
-
-        if (noMerge) {
-          metadata.noMerge = true
-        } else {
-          try {
-            await this.pgQx.tx(async (txPgQx) => {
-              const service = new CommonMemberService(txPgQx, this.temporal, this.log)
-              await service.merge(originalId, targetId)
-            })
-
+        try {
+          const merged = await mergeIfAllowed(
+            this.pgQx,
+            this.temporal,
+            this.log,
+            originalId,
+            targetId,
+          )
+          if (merged) {
             return originalId
-          } catch (err) {
-            metadata.mergeError = {
-              errorMessage: err?.message ?? '<no error message>',
-              errorStack: err?.stack,
-              err,
-            }
+          } else {
+            metadata.noMerge = true
+          }
+        } catch (err) {
+          metadata.mergeError = {
+            errorMessage: err?.message ?? '<no error message>',
+            errorStack: err?.stack,
+            err,
           }
         }
       }