Managing activity relations in a separate table

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/V1737705711__activityRelations.sql

@@ -0,0 +1,55 @@
+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", "objectMemberId", "organizationId", "conversationId", "parentId", "platform", "username", "objectMemberUsername")
+        SELECT id, "memberId", "objectMemberId", "organizationId", "conversationId", "parentId", "platform", "username", "objectMemberUsername"
+        FROM activities
+        WHERE id > last_processed_id
+        ORDER BY id
+        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 MAX(id) INTO last_processed_id FROM activities WHERE id > last_processed_id;
+
+    END LOOP;
+
+    RAISE NOTICE 'All rows processed. Total rows inserted: %.', total_processed;
+END
+$$;

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

@@ -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/services/member/memberAffiliationsService.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'
@@ -13,6 +13,7 @@ import {
 
 import MemberAffiliationsRepository from '@/database/repositories/member/memberAffiliationsRepository'
 import MemberOrganizationAffiliationOverridesRepository from '@/database/repositories/member/memberOrganizationAffiliationOverridesRepository'
+import MemberOrganizationsRepository from '@/database/repositories/member/memberOrganizationsRepository'
 import SequelizeRepository from '@/database/repositories/sequelizeRepository'
 
 import { IServiceOptions } from '../IServiceOptions'
@@ -66,6 +67,40 @@ export default class MemberAffiliationsService extends LoggerBase {
   async changeAffiliationOverride(
     data: IChangeAffiliationOverrideData,
   ): Promise<IMemberOrganizationAffiliationOverride> {
+    if (data.isPrimaryWorkExperience) {
+      // 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 MemberOrganizationsRepository.list(data.memberId, this.options)
+      ).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,

services/apps/data_sink_worker/src/service/activity.service.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

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

@@ -1,5 +1,9 @@
 import { WorkflowIdReusePolicy } from '@temporalio/workflow'
 
+import {
+  moveActivityRelationsToAnotherMember,
+  moveActivityRelationsWithIdentityToAnotherMember,
+} from '@crowd/data-access-layer'
 import { cleanupMemberAggregates } from '@crowd/data-access-layer/src/members/segments'
 import {
   cleanupMember,
@@ -39,6 +43,7 @@ export async function moveActivitiesBetweenMembers(
     return
   }
   await moveActivitiesToNewMember(svc.questdbSQL, svc.queue, primaryId, secondaryId, tenantId)
+  await moveActivityRelationsToAnotherMember(dbStoreQx(svc.postgres.writer), secondaryId, primaryId)
 }
 
 export async function moveActivitiesWithIdentityToAnotherMember(
@@ -74,6 +79,13 @@ export async function moveActivitiesWithIdentityToAnotherMember(
       identity.value,
       identity.platform,
     )
+    await moveActivityRelationsWithIdentityToAnotherMember(
+      dbStoreQx(svc.postgres.writer),
+      fromId,
+      toId,
+      identity.value,
+      identity.platform,
+    )
   }
 }
 

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

@@ -1,3 +1,4 @@
+import { moveActivityRelationsToAnotherOrganization } from '@crowd/data-access-layer'
 import {
   deleteOrganizationById,
   deleteOrganizationSegments,
@@ -30,6 +31,11 @@ export async function moveActivitiesBetweenOrgs(
   tenantId: string,
 ): Promise<void> {
   await moveActivitiesToNewOrg(svc.questdbSQL, svc.queue, primaryId, secondaryId, tenantId)
+  await moveActivityRelationsToAnotherOrganization(
+    dbStoreQx(svc.postgres.writer),
+    primaryId,
+    secondaryId,
+  )
 }
 
 export async function recalculateActivityAffiliationsOfOrganizationSynchronous(

services/libs/common/src/timing.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
+}