Merge branch 'main' into support/merge-error

Author: skwowet

backend/package.json

@@ -36,7 +36,7 @@
     "script:refreshGithubRepoSettings": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true tsx src/bin/scripts/refresh-github-repo-settings.ts"
   },
   "lint-staged": {
-    "**/*.*": [
+    "**/*.ts": [
       "eslint",
       "prettier --write"
     ]

backend/src/database/migrations/U1737705711__activityRelations.sql

@@ -0,0 +1,60 @@
+create table public."activityRelations" (
+    "activityId" uuid not null primary key,
+    "memberId" uuid not null,
+    "objectMemberId" uuid null,
+    "organizationId" uuid null,
+    "conversationId" uuid null,
+    "parentId" uuid null,
+    "segmentId" uuid not null,
+    "platform" text not null,
+    "username" text not null,
+    "objectMemberUsername" text null,
+    "createdAt" timestamp with time zone default now() not null,
+    "updatedAt" timestamp with time zone default now() not null,
+    foreign key ("memberId") references members (id) on delete cascade,
+    foreign key ("organizationId") references organizations (id) on delete set null,
+    foreign key ("objectMemberId") references members (id) on delete set null,
+    foreign key ("conversationId") references conversations (id) on delete set null,
+    foreign key ("segmentId") references segments (id) on delete cascade,
+    unique ("activityId", "memberId")
+);
+create index "ix_activityRelations_memberId" on "activityRelations"("memberId");
+create index "ix_activityRelations_organizationId" on "activityRelations"("organizationId");
+create index "ix_activityRelations_platform_username" on "activityRelations"("platform", "username");
+
+
+DO
+$$
+DECLARE
+    batch_size INT := 100000;
+    last_processed_id UUID := '00000000-0000-0000-0000-000000000000';
+    total_processed INT := 0;
+    rows_inserted INT;
+BEGIN
+    LOOP
+        INSERT INTO "activityRelations" ("activityId", "memberId", "segmentId", "objectMemberId", "organizationId", "conversationId", "parentId", "platform", "username", "objectMemberUsername")
+        SELECT id, "memberId", "segmentId", "objectMemberId", "organizationId", "conversationId", "parentId", "platform", "username", "objectMemberUsername"
+        FROM activities
+        WHERE id > last_processed_id
+        AND "segmentId" IS NOT NULL
+        ORDER BY id::TEXT
+        LIMIT batch_size;
+
+        GET DIAGNOSTICS rows_inserted = ROW_COUNT;
+
+        total_processed := total_processed + rows_inserted;
+        RAISE NOTICE 'Batch processed: % rows. Total processed: % rows.', rows_inserted, total_processed;
+
+        EXIT WHEN rows_inserted = 0;
+
+        SELECT id::UUID INTO last_processed_id
+        FROM activities
+        WHERE id > last_processed_id
+        ORDER BY id::TEXT DESC
+        LIMIT 1;
+
+    END LOOP;
+
+    RAISE NOTICE 'All rows processed. Total rows inserted: %.', total_processed;
+END
+$$;

backend/src/database/migrations/V1737705711__activityRelations.sql

@@ -1,8 +1,12 @@
 import {
   changeOverride as changeMemberOrganizationAffiliationOverride,
   findOverrides as findMemberOrganizationAffiliationOverrides,
+  findPrimaryWorkExperiencesOfMember,
 } from '@crowd/data-access-layer/src/member_organization_affiliation_overrides'
-import { IChangeAffiliationOverrideData } from '@crowd/types'
+import {
+  IChangeAffiliationOverrideData,
+  IMemberOrganizationAffiliationOverride,
+} from '@crowd/types'
 
 import { IRepositoryOptions } from '../IRepositoryOptions'
 import SequelizeRepository from '../sequelizeRepository'
@@ -17,6 +21,14 @@ class MemberOrganizationAffiliationOverridesRepository {
     ])
     return overrides[0]
   }
+
+  static async findPrimaryWorkExperiences(
+    memberId: string,
+    options: IRepositoryOptions,
+  ): Promise<IMemberOrganizationAffiliationOverride[]> {
+    const qx = SequelizeRepository.getQueryExecutor(options)
+    return findPrimaryWorkExperiencesOfMember(qx, memberId)
+  }
 }
 
 export default MemberOrganizationAffiliationOverridesRepository

backend/src/database/repositories/member/memberOrganizationAffiliationOverridesRepository.ts

@@ -1,7 +1,7 @@
 /* eslint-disable no-continue */
 import { uniq } from 'lodash'
 
-import { groupBy } from '@crowd/common'
+import { Error400, dateIntersects, groupBy } from '@crowd/common'
 import { findMaintainerRoles } from '@crowd/data-access-layer/src/maintainers'
 import { fetchManySegments } from '@crowd/data-access-layer/src/segments'
 import { LoggerBase } from '@crowd/logging'
@@ -18,6 +18,8 @@ import SequelizeRepository from '@/database/repositories/sequelizeRepository'
 import { IServiceOptions } from '../IServiceOptions'
 import MemberAffiliationService from '../memberAffiliationService'
 
+import MemberOrganizationsService from './memberOrganizationsService'
+
 export default class MemberAffiliationsService extends LoggerBase {
   options: IServiceOptions
 
@@ -66,6 +68,41 @@ export default class MemberAffiliationsService extends LoggerBase {
   async changeAffiliationOverride(
     data: IChangeAffiliationOverrideData,
   ): Promise<IMemberOrganizationAffiliationOverride> {
+    if (data.isPrimaryWorkExperience) {
+      const memberOrgService = new MemberOrganizationsService(this.options)
+      // check if any other work experience in intersecting date range was marked as primary
+      // we don't allow this because "isPrimaryWorkExperience" decides which work exp to pick on date conflicts
+      const allWorkExperiencesOfMember = (await memberOrgService.list(data.memberId)).map(
+        (mo) => mo.memberOrganizations,
+      )
+
+      const currentlyEditedWorkExperience = allWorkExperiencesOfMember.find(
+        (w) => w.id === data.memberOrganizationId,
+      )
+
+      const primaryWorkExperiencesOfMember = allWorkExperiencesOfMember.filter(
+        (w) => w.affiliationOverride.isPrimaryWorkExperience,
+      )
+
+      if (currentlyEditedWorkExperience.affiliationOverride.isPrimaryWorkExperience === false) {
+        for (const existingPrimaryWorkExp of primaryWorkExperiencesOfMember) {
+          if (
+            dateIntersects(
+              existingPrimaryWorkExp.dateStart as string,
+              existingPrimaryWorkExp.dateEnd as string,
+              currentlyEditedWorkExperience.dateStart as string,
+              currentlyEditedWorkExperience.dateEnd as string,
+            )
+          ) {
+            throw new Error400(
+              this.options.language,
+              `Date range conflicts with another primary work experience id = ${existingPrimaryWorkExp.id}`,
+            )
+          }
+        }
+      }
+    }
+
     const override = MemberOrganizationAffiliationOverridesRepository.changeOverride(
       data,
       this.options,

backend/src/services/member/memberAffiliationsService.ts

@@ -12,6 +12,7 @@ import {
 } from '@crowd/common'
 import { SearchSyncWorkerEmitter } from '@crowd/common_services'
 import {
+  createOrUpdateRelations,
   findCommitsForPRSha,
   findMatchingPullRequestNodeId,
   insertActivities,
@@ -31,6 +32,7 @@ import { IDbMember } from '@crowd/data-access-layer/src/old/apps/data_sink_worke
 import MemberRepository from '@crowd/data-access-layer/src/old/apps/data_sink_worker/repo/member.repo'
 import RequestedForErasureMemberIdentitiesRepository from '@crowd/data-access-layer/src/old/apps/data_sink_worker/repo/requestedForErasureMemberIdentities.repo'
 import SettingsRepository from '@crowd/data-access-layer/src/old/apps/data_sink_worker/repo/settings.repo'
+import { dbStoreQx } from '@crowd/data-access-layer/src/queryExecutor'
 import { DEFAULT_ACTIVITY_TYPE_SETTINGS, GithubActivityType } from '@crowd/integrations'
 import { GitActivityType } from '@crowd/integrations/src/integrations/git/types'
 import { Logger, LoggerBase, getChildLogger } from '@crowd/logging'
@@ -84,6 +86,7 @@ export default class ActivityService extends LoggerBase {
       })
 
       const id = await this.pgStore.transactionally(async (txStore) => {
+        const queryExecutor = dbStoreQx(txStore)
         const txSettingsRepo = new SettingsRepository(txStore, this.log)
 
         await txSettingsRepo.createActivityType(
@@ -132,6 +135,16 @@ export default class ActivityService extends LoggerBase {
               importHash: activity.importHash,
             },
           ])
+          await createOrUpdateRelations(queryExecutor, {
+            activityId: activity.id,
+            segmentId,
+            memberId: activity.memberId,
+            objectMemberId: activity.objectMemberId,
+            organizationId: activity.organizationId,
+            platform: activity.platform,
+            username: activity.username,
+            objectMemberUsername: activity.objectMemberUsername,
+          })
         } catch (error) {
           this.log.error('Error creating activity in QuestDB:', error)
           throw error
@@ -160,6 +173,7 @@ export default class ActivityService extends LoggerBase {
     try {
       let toUpdate: IDbActivityUpdateData
       const updated = await this.pgStore.transactionally(async (txStore) => {
+        const queryExecutor = dbStoreQx(txStore)
         const txSettingsRepo = new SettingsRepository(txStore, this.log)
 
         toUpdate = await this.mergeActivityData(activity, original)
@@ -215,6 +229,16 @@ export default class ActivityService extends LoggerBase {
                 importHash: original.importHash,
               },
             ])
+            await createOrUpdateRelations(queryExecutor, {
+              activityId: id,
+              segmentId,
+              memberId: toUpdate.memberId || original.memberId,
+              objectMemberId: toUpdate.objectMemberId || original.objectMemberId,
+              organizationId: toUpdate.organizationId || original.organizationId,
+              platform: toUpdate.platform || (original.platform as PlatformType),
+              username: toUpdate.username || original.username,
+              objectMemberUsername: toUpdate.objectMemberUsername || original.objectMemberUsername,
+            })
           } catch (error) {
             this.log.error('Error updating (by inserting) activity in QuestDB:', error)
             throw error
@@ -1338,6 +1362,7 @@ export default class ActivityService extends LoggerBase {
     }) => {
       await updateActivities(
         this.qdbStore.connection(),
+        dbStoreQx(this.pgStore),
         this.client,
         async (activity) => ({
           attributes: {
@@ -1388,6 +1413,7 @@ export default class ActivityService extends LoggerBase {
 
     await updateActivities(
       this.qdbStore.connection(),
+      dbStoreQx(this.pgStore),
       this.client,
       async () => ({
         sourceParentId: activity.sourceId,

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

@@ -1,6 +1,5 @@
 export {
   deleteMember,
-  moveActivitiesWithIdentityToAnotherMember,
   recalculateActivityAffiliationsOfMemberAsync,
   syncMember,
   notifyFrontendMemberMergeSuccessful,

services/apps/entity_merging_worker/src/activities.ts

@@ -6,9 +6,6 @@ import { IDbActivityCreateData } from '@crowd/data-access-layer/src/old/apps/dat
 import {
   cleanupMember,
   deleteMemberSegments,
-  findMemberById,
-  getIdentitiesWithActivity,
-  moveIdentityActivitiesToNewMember,
 } from '@crowd/data-access-layer/src/old/apps/entity_merging_worker'
 import { figureOutNewOrgId } from '@crowd/data-access-layer/src/old/apps/profiles_worker'
 import { prepareMemberAffiliationsUpdate } from '@crowd/data-access-layer/src/old/apps/profiles_worker'
@@ -31,42 +28,6 @@ export async function deleteMember(memberId: string): Promise<void> {
   await cleanupMember(svc.postgres.writer, memberId)
 }
 
-export async function moveActivitiesWithIdentityToAnotherMember(
-  fromId: string,
-  toId: string,
-  identities: IMemberIdentity[],
-  tenantId: string,
-): Promise<void> {
-  const memberExists = await findMemberById(svc.postgres.writer, toId, tenantId)
-
-  if (!memberExists) {
-    return
-  }
-
-  const identitiesWithActivity = await getIdentitiesWithActivity(
-    svc.postgres.writer,
-    fromId,
-    tenantId,
-    identities,
-  )
-
-  for (const identity of identities.filter(
-    (i) =>
-      i.type === MemberIdentityType.USERNAME &&
-      identitiesWithActivity.some((ai) => ai.platform === i.platform && ai.username === i.value),
-  )) {
-    await moveIdentityActivitiesToNewMember(
-      svc.questdbSQL,
-      svc.queue,
-      tenantId,
-      fromId,
-      toId,
-      identity.value,
-      identity.platform,
-    )
-  }
-}
-
 export async function recalculateActivityAffiliationsOfMemberAsync(
   memberId: string,
   tenantId: string,
@@ -184,14 +145,17 @@ export async function finishMemberMergingUpdateActivities(memberId: string, newM
   const queueClient = svc.queue
 
   const qx = pgpQx(pgDb.connection())
-  const { orgCases } = await prepareMemberAffiliationsUpdate(qx, memberId)
+  const { orgCases, fallbackOrganizationId } = await prepareMemberAffiliationsUpdate(qx, memberId)
 
   await updateActivities(
     qDb,
+    pgpQx(svc.postgres.writer.connection()),
     queueClient,
     [
       async () => ({ memberId: newMemberId }),
-      async (activity) => ({ organizationId: figureOutNewOrgId(activity, orgCases) }),
+      async (activity) => ({
+        organizationId: figureOutNewOrgId(activity, orgCases, fallbackOrganizationId),
+      }),
     ],
     `"memberId" = $(memberId)`,
     {
@@ -232,15 +196,16 @@ export async function finishMemberUnmergingUpdateActivities({
   const queueClient = svc.queue
 
   const qx = pgpQx(pgDb.connection())
-  const { orgCases } = await prepareMemberAffiliationsUpdate(qx, memberId)
+  const { orgCases, fallbackOrganizationId } = await prepareMemberAffiliationsUpdate(qx, memberId)
 
   await updateActivities(
     qDb,
+    pgpQx(svc.postgres.writer.connection()),
     queueClient,
     [
       async (activity) => moveByIdentities({ activity, identities, newMemberId }),
       async (activity) => ({
-        organizationId: figureOutNewOrgId(activity, orgCases),
+        organizationId: figureOutNewOrgId(activity, orgCases, fallbackOrganizationId),
       }),
     ],
     `"memberId" = $(memberId)`,

services/apps/entity_merging_worker/src/activities/members.ts

@@ -29,7 +29,14 @@ export async function moveActivitiesBetweenOrgs(
   secondaryId: string,
   tenantId: string,
 ): Promise<void> {
-  await moveActivitiesToNewOrg(svc.questdbSQL, svc.queue, primaryId, secondaryId, tenantId)
+  await moveActivitiesToNewOrg(
+    svc.questdbSQL,
+    svc.postgres.writer.connection(),
+    svc.queue,
+    primaryId,
+    secondaryId,
+    tenantId,
+  )
 }
 
 export async function recalculateActivityAffiliationsOfOrganizationSynchronous(

services/apps/entity_merging_worker/src/activities/organizations.ts

@@ -57,3 +57,25 @@ export const getLongestDateRange = <T extends IMemberOrganization>(orgs: T[]) =>
 
   return sortedByDateRange[0]
 }
+
+export const dateIntersects = (
+  d1Start?: string | null,
+  d1End?: string | null,
+  d2Start?: string | null,
+  d2End?: string | null,
+): boolean => {
+  // If both periods have no dates at all, we consider they span all time
+  if ((!d1Start && !d1End) || (!d2Start && !d2End)) {
+    return true
+  }
+
+  // Convert strings to timestamps, using fallbacks for missing dates
+  const start1 = d1Start ? new Date(d1Start).getTime() : -Infinity
+  const end1 = d1End ? new Date(d1End).getTime() : Infinity
+  const start2 = d2Start ? new Date(d2Start).getTime() : -Infinity
+  const end2 = d2End ? new Date(d2End).getTime() : Infinity
+
+  // Periods intersect if one period's start is before other period's end
+  // and that same period's end is after the other period's start
+  return start1 <= end2 && end1 >= start2
+}