From 97e6711032b2507b0291619c45531aeed3d5b0d9 Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Fri, 7 Aug 2026 12:05:53 +0200 Subject: [PATCH 1/8] fix: filter soft-deleted Signed-off-by: Umberto Sgueglia --- .../src/activities/enrichment.ts | 7 ++++++- .../src/old/apps/members_enrichment_worker/index.ts | 13 +++++++++++-- services/libs/types/src/enrichment.ts | 6 ++++++ 3 files changed, 23 insertions(+), 3 deletions(-) diff --git a/services/apps/members_enrichment_worker/src/activities/enrichment.ts b/services/apps/members_enrichment_worker/src/activities/enrichment.ts index b646d5a759..81693be92a 100644 --- a/services/apps/members_enrichment_worker/src/activities/enrichment.ts +++ b/services/apps/members_enrichment_worker/src/activities/enrichment.ts @@ -244,7 +244,7 @@ export async function getPriorityArray(): Promise { export async function fetchMemberDataForLLMSquashing( memberId: string, -): Promise { +): Promise { return fetchMemberDataForLLMSquashingDb(svc.postgres.reader.connection(), memberId) } @@ -601,6 +601,7 @@ export async function updateMemberUsingSquashedPayload( existingMemberData.organizations, squashedPayload.memberOrganizations, isHighConfidenceSourceSelectedForWorkExperiences, + new Set(existingMemberData.deletedOrganizations.map((o) => o.orgId)), ) // Enrichment often deletes and recreates the same orgs with identical dates. @@ -834,12 +835,16 @@ function prepareWorkExperiences( oldVersion: IMemberOrganizationData[], newVersion: IMemberEnrichmentDataNormalizedOrganization[], isHighConfidenceSourceSelectedForWorkExperiences: boolean, + deletedOrganizationIds: Set, ): IWorkExperienceChanges { // we delete all the work experiences that were not manually created or from the project registry. const toDelete = oldVersion.filter( (c) => c.source !== OrganizationSource.UI && c.source !== OrganizationSource.PROJECT_REGISTRY, ) + // never recreate an affiliation that was manually deleted — enrichment providers keep resupplying it + newVersion = newVersion.filter((e) => !deletedOrganizationIds.has(e.organizationId)) + const toCreate: IMemberEnrichmentDataNormalizedOrganization[] = [] // eslint-disable-next-line @typescript-eslint/no-explicit-any const toUpdate: Map> = new Map() diff --git a/services/libs/data-access-layer/src/old/apps/members_enrichment_worker/index.ts b/services/libs/data-access-layer/src/old/apps/members_enrichment_worker/index.ts index 51ffd36fc5..e6421d433e 100644 --- a/services/libs/data-access-layer/src/old/apps/members_enrichment_worker/index.ts +++ b/services/libs/data-access-layer/src/old/apps/members_enrichment_worker/index.ts @@ -57,7 +57,13 @@ export async function fetchMemberDataForLLMSquashing( where mo."memberId" = $(memberId) and mo."deletedAt" is null and o."deletedAt" is null - group by mo."memberId", mo."organizationId", o."displayName", mo.id) + group by mo."memberId", mo."organizationId", o."displayName", mo.id), + deleted_member_orgs as (select distinct + mo."organizationId" as "orgId" + from "memberOrganizations" mo + where mo."memberId" = $(memberId) + and mo."deletedAt" is not null + and mo.source not in ('ui', 'project-registry')) select m."displayName", m.attributes, m."manuallyChangedFields", @@ -90,7 +96,10 @@ export async function fetchMemberDataForLLMSquashing( where mo."memberId" = m.id ) else '[]'::json - end as organizations + end as organizations, + coalesce( + (select json_agg(jsonb_build_object('orgId', d."orgId") order by d."orgId") from deleted_member_orgs d), '[]'::json + ) as "deletedOrganizations" from members m where m.id = $(memberId) and m."deletedAt" is null diff --git a/services/libs/types/src/enrichment.ts b/services/libs/types/src/enrichment.ts index 831de9d529..e1b21d95a4 100644 --- a/services/libs/types/src/enrichment.ts +++ b/services/libs/types/src/enrichment.ts @@ -43,6 +43,10 @@ export interface IMemberOrganizationData { identities?: IOrganizationIdentity[] } +export interface IDeletedMemberOrganizationData { + orgId: string +} + export interface IMemberOriginalData { // members table data displayName: string @@ -55,6 +59,8 @@ export interface IMemberOriginalData { // memberOrganizations table data organizations: IMemberOrganizationData[] + // memberOrganizations rows manually deleted, source not UI/PROJECT_REGISTRY — tombstones enrichment must not recreate + deletedOrganizations: IDeletedMemberOrganizationData[] } export interface IOrganizationEnrichmentCache { From d21e70536b1bf5bd4a36c31a10b9d47b2b96dfaa Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Thu, 13 Aug 2026 19:17:35 +0200 Subject: [PATCH 2/8] fix: update-model store deletedBy Signed-off-by: Umberto Sgueglia --- .../verifyMemberWorkExperience.ts | 5 +- ...add-deleted-by-to-member-organizations.sql | 2 + .../member/memberOrganizationsService.ts | 8 +- pnpm-lock.yaml | 3 + .../members_enrichment_worker/package.json | 6 +- .../src/activities/enrichment.ts | 155 +------------- .../workExperienceReconciliation.test.ts | 186 +++++++++++++++++ .../workExperienceReconciliation.ts | 197 ++++++++++++++++++ .../vitest.config.ts | 13 ++ .../src/members/organizations.ts | 11 +- .../apps/members_enrichment_worker/index.ts | 8 +- services/libs/types/src/enrichment.ts | 4 +- 12 files changed, 443 insertions(+), 155 deletions(-) create mode 100644 backend/src/database/migrations/V1786621970__add-deleted-by-to-member-organizations.sql create mode 100644 services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.test.ts create mode 100644 services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.ts create mode 100644 services/apps/members_enrichment_worker/vitest.config.ts diff --git a/backend/src/api/public/v1/members/work-experiences/verifyMemberWorkExperience.ts b/backend/src/api/public/v1/members/work-experiences/verifyMemberWorkExperience.ts index 0491bf7079..ed2b1d975e 100644 --- a/backend/src/api/public/v1/members/work-experiences/verifyMemberWorkExperience.ts +++ b/backend/src/api/public/v1/members/work-experiences/verifyMemberWorkExperience.ts @@ -92,8 +92,9 @@ export async function verifyMemberWorkExperience(req: Request, res: Response): P await updateMemberOrganization(tx, memberId, overlappingRow.id, verifiedUpdate) } } else { - // Unverifying removes the grouped work experience from both visible and hidden rows - await deleteMemberOrganizations(tx, memberId, memberOrgIdsToDelete, true) + // Unverifying removes the grouped work experience from both visible and hidden rows. + // This is a human decision, so deletedBy is set — enrichment must never recreate it. + await deleteMemberOrganizations(tx, memberId, memberOrgIdsToDelete, true, verifiedBy) } }) diff --git a/backend/src/database/migrations/V1786621970__add-deleted-by-to-member-organizations.sql b/backend/src/database/migrations/V1786621970__add-deleted-by-to-member-organizations.sql new file mode 100644 index 0000000000..382cd32508 --- /dev/null +++ b/backend/src/database/migrations/V1786621970__add-deleted-by-to-member-organizations.sql @@ -0,0 +1,2 @@ +alter table "memberOrganizations" + add column if not exists "deletedBy" varchar(255) default null; diff --git a/backend/src/services/member/memberOrganizationsService.ts b/backend/src/services/member/memberOrganizationsService.ts index 450ce65876..bec257d0e4 100644 --- a/backend/src/services/member/memberOrganizationsService.ts +++ b/backend/src/services/member/memberOrganizationsService.ts @@ -363,7 +363,13 @@ export default class MemberOrganizationsService extends LoggerBase { ] // Delete hidden grouped rows with the visible row so list responses stay consistent - await deleteMemberOrganizations(qx, memberId, memberOrganizationIdsToDelete, true) + await deleteMemberOrganizations( + qx, + memberId, + memberOrganizationIdsToDelete, + true, + this.options.currentUser.id, + ) const result = await this.list(memberId, transaction) diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 44e00af382..dfe2aaf8db 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1071,6 +1071,9 @@ importers: nodemon: specifier: ^3.0.1 version: 3.1.0 + vitest: + specifier: ^3.2.4 + version: 3.2.4(@types/debug@4.1.12)(@types/node@20.12.7)(jiti@2.4.2)(terser@5.43.1)(tsx@4.7.3)(yaml@2.7.0) services/apps/merge_suggestions_worker: dependencies: diff --git a/services/apps/members_enrichment_worker/package.json b/services/apps/members_enrichment_worker/package.json index 64da9ef784..d1446827fc 100644 --- a/services/apps/members_enrichment_worker/package.json +++ b/services/apps/members_enrichment_worker/package.json @@ -10,7 +10,8 @@ "lint": "npx eslint --ext .ts src --max-warnings=0", "format": "npx prettier --write \"src/**/*.ts\"", "format-check": "npx prettier --check .", - "tsc-check": "tsc --noEmit" + "tsc-check": "tsc --noEmit", + "test": "vitest run" }, "dependencies": { "@crowd/archetype-standard": "workspace:*", @@ -35,6 +36,7 @@ "devDependencies": { "@types/node": "^20.8.2", "@types/uuid": "~9.0.6", - "nodemon": "^3.0.1" + "nodemon": "^3.0.1", + "vitest": "^3.2.4" } } diff --git a/services/apps/members_enrichment_worker/src/activities/enrichment.ts b/services/apps/members_enrichment_worker/src/activities/enrichment.ts index 81693be92a..4b03e23611 100644 --- a/services/apps/members_enrichment_worker/src/activities/enrichment.ts +++ b/services/apps/members_enrichment_worker/src/activities/enrichment.ts @@ -3,7 +3,6 @@ import axios from 'axios' import _ from 'lodash' import { - generateUUIDv1, getAttributeValue, getCountry, hasAttributeValue, @@ -60,7 +59,6 @@ import { OrganizationAttributeSource, OrganizationIdentityType, OrganizationMergeSuggestionTable, - OrganizationSource, PlatformType, } from '@crowd/types' @@ -74,6 +72,11 @@ import { IMemberEnrichmentDataNormalizedOrganization, } from '../types' +import { + hasMemberOrganizationTimelineChange, + prepareWorkExperiences, +} from './workExperienceReconciliation' + /* eslint-disable @typescript-eslint/no-explicit-any */ // Get the most strict parallelism among existing and enrichable sources @@ -604,10 +607,13 @@ export async function updateMemberUsingSquashedPayload( new Set(existingMemberData.deletedOrganizations.map((o) => o.orgId)), ) - // Enrichment often deletes and recreates the same orgs with identical dates. - // Skip the refresh when the timeline that drives activityRelations hasn't changed. + // Skip the refresh when the timeline that drives activityRelations hasn't changed — + // e.g. a title-only update-in-place shouldn't trigger a full recompute. + const toUpdateHasTimelineChange = Array.from(results.toUpdate.values()).some( + (fields) => 'dateStart' in fields || 'dateEnd' in fields, + ) affiliationNeedsRefresh = - results.toUpdate.size > 0 || + toUpdateHasTimelineChange || hasMemberOrganizationTimelineChange(results.toDelete, results.toCreate) if (results.toDelete.length > 0) { @@ -789,12 +795,6 @@ export async function refreshMemberEnrichmentMaterializedView(mvName: string): P await refreshMaterializedView(svc.postgres.writer.connection(), mvName, true) } -interface IWorkExperienceChanges { - toDelete: IMemberOrganizationData[] - toCreate: IMemberEnrichmentDataNormalizedOrganization[] - toUpdate: Map> -} - function sanitizeWorkExperienceDateRanges( organizations: IMemberEnrichmentDataNormalizedOrganization[], ): IMemberEnrichmentDataNormalizedOrganization[] { @@ -809,139 +809,6 @@ function sanitizeWorkExperienceDateRanges( }) } -/** - * Returns true when the set of (orgId, startDate, endDate) tuples differs - * between deletes and creates. Fields like title or source don't affect - * the affiliation timeline, so they're intentionally ignored. - */ -function hasMemberOrganizationTimelineChange( - toDelete: IMemberOrganizationData[], - toCreate: IMemberEnrichmentDataNormalizedOrganization[], -): boolean { - const toKey = (orgId: string, start: string | null | undefined, end: string | null | undefined) => - `${orgId}|${start ? start.substring(0, 10) : ''}|${end ? end.substring(0, 10) : ''}` - - const deletedKeys = new Set(toDelete.map((d) => toKey(d.orgId, d.dateStart, d.dateEnd))) - const createdKeys = new Set(toCreate.map((c) => toKey(c.organizationId, c.startDate, c.endDate))) - - if (deletedKeys.size !== createdKeys.size) return true - for (const key of deletedKeys) { - if (!createdKeys.has(key)) return true - } - return false -} - -function prepareWorkExperiences( - oldVersion: IMemberOrganizationData[], - newVersion: IMemberEnrichmentDataNormalizedOrganization[], - isHighConfidenceSourceSelectedForWorkExperiences: boolean, - deletedOrganizationIds: Set, -): IWorkExperienceChanges { - // we delete all the work experiences that were not manually created or from the project registry. - const toDelete = oldVersion.filter( - (c) => c.source !== OrganizationSource.UI && c.source !== OrganizationSource.PROJECT_REGISTRY, - ) - - // never recreate an affiliation that was manually deleted — enrichment providers keep resupplying it - newVersion = newVersion.filter((e) => !deletedOrganizationIds.has(e.organizationId)) - - const toCreate: IMemberEnrichmentDataNormalizedOrganization[] = [] - // eslint-disable-next-line @typescript-eslint/no-explicit-any - const toUpdate: Map> = new Map() - - if (isHighConfidenceSourceSelectedForWorkExperiences) { - const uiEntries = oldVersion.filter((c) => c.source === OrganizationSource.UI) - const filteredNewVersion = newVersion.filter( - (e) => - !uiEntries.some( - (ui) => - e.title === ui.jobTitle && - e.identities && - e.identities.some((i) => i.organizationId === ui.orgId), - ), - ) - toCreate.push(...filteredNewVersion) - return { - toDelete, - toCreate, - toUpdate, - } - } - - // sort both versions by start date and only use manual changes from the current version - const orderedCurrentVersion = oldVersion - .filter((c) => c.source === OrganizationSource.UI) - .sort((a, b) => { - // If either value is null/undefined, move it to the beginning - if (!a.dateStart && !b.dateStart) return 0 - if (!a.dateStart) return -1 - if (!b.dateStart) return 1 - - // Compare dates if both values exist - return new Date(a.dateStart as string).getTime() - new Date(b.dateStart as string).getTime() - }) - - let orderedNewVersion = newVersion.sort((a, b) => { - // If either value is null/undefined, move it to the beginning - if (!a.startDate && !b.startDate) return 0 - if (!a.startDate) return -1 - if (!b.startDate) return 1 - - // Compare dates if both values exist - return new Date(a.startDate as string).getTime() - new Date(b.startDate as string).getTime() - }) - - // set ids and new flag to new versions just so we can easily manipulate the array later - for (const exp of orderedNewVersion) { - exp.id = generateUUIDv1() - } - - // we iterate through the existing version experiences to see if update is needed - for (const current of orderedCurrentVersion) { - // try and find a matching experience in the new versions by title - const match = orderedNewVersion.find( - (e) => - e.title === current.jobTitle && - e.identities && - e.identities.some((e) => e.organizationId === current.orgId), - ) - - // if we found a match we can check if we need something to update - if ( - match && - current.dateStart === match.startDate && - current.dateEnd === null && - match.endDate !== null - ) { - const toUpdateInner: Record = {} - - toUpdateInner.dateEnd = match.endDate - toUpdate.set(current, toUpdateInner) - - // remove the match from the new version array so we later don't process it again - orderedNewVersion = orderedNewVersion.filter((e) => e.id !== match.id) - } else if ( - match && - (current.dateStart !== match.startDate || current.dateEnd !== null || match.endDate === null) - ) { - // there's an incoming work experiences, but it's conflicting with the existing manually updated data - // we shouldn't add or update anything when this happens - // we can only update dateEnd of existing manually changed data, when it has a null dateEnd - orderedNewVersion = orderedNewVersion.filter((e) => e.id !== match.id) - } - // if we didn't find a match we should just leave it as it is in the database since it was manual input - } - - // the remaining experiences in the new version array are just new experiences to create - toCreate.push(...orderedNewVersion) - - return { - toDelete, - toCreate, - toUpdate, - } -} - export async function syncMember(memberId: string): Promise { const syncApi = new SearchSyncApiClient({ baseUrl: process.env['CROWD_SEARCH_SYNC_API_URL'], diff --git a/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.test.ts b/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.test.ts new file mode 100644 index 0000000000..b0664688d7 --- /dev/null +++ b/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.test.ts @@ -0,0 +1,186 @@ +import { describe, expect, it } from 'vitest' + +import { IMemberOrganizationData, OrganizationIdentityType, OrganizationSource } from '@crowd/types' + +import { IMemberEnrichmentDataNormalizedOrganization } from '../types' + +import { + hasMemberOrganizationTimelineChange, + prepareWorkExperiences, +} from './workExperienceReconciliation' + +function oldRow(overrides: Partial = {}): IMemberOrganizationData { + return { + id: 'row-1', + orgId: 'org-1', + jobTitle: 'Engineer', + dateStart: '2020-01-01', + dateEnd: null, + source: OrganizationSource.ENRICHMENT_PROGAI, + verified: false, + verifiedBy: null, + ...overrides, + } +} + +function newEntry( + overrides: Partial = {}, +): IMemberEnrichmentDataNormalizedOrganization { + const organizationId = overrides.organizationId ?? 'org-1' + return { + organizationId, + name: 'Org One', + title: 'Engineer', + startDate: '2020-01-01', + endDate: null, + source: OrganizationSource.ENRICHMENT_PROGAI, + identities: [ + { + organizationId, + platform: 'linkedin', + value: 'org-one', + type: OrganizationIdentityType.USERNAME, + verified: true, + }, + ], + ...overrides, + } +} + +describe('prepareWorkExperiences', () => { + it('leaves a verified row untouched when the provider resupplies a matching entry', () => { + const verified = oldRow({ id: 'row-verified', verified: true, verifiedBy: 'jane' }) + const result = prepareWorkExperiences([verified], [newEntry()], false, new Set()) + + expect(result.toDelete).not.toContain(verified) + expect(result.toUpdate.has(verified)).toBe(false) + }) + + it('leaves a verified row untouched even when the provider sends conflicting dates', () => { + const verified = oldRow({ id: 'row-verified', verified: true, verifiedBy: 'jane' }) + const conflicting = newEntry({ startDate: '2021-06-01', endDate: '2022-01-01' }) + const result = prepareWorkExperiences([verified], [conflicting], false, new Set()) + + expect(result.toDelete).not.toContain(verified) + expect(result.toUpdate.has(verified)).toBe(false) + }) + + it('never recreates an organization a person deleted on purpose (tombstoned)', () => { + const tombstonedEntry = newEntry({ organizationId: 'org-deleted' }) + const result = prepareWorkExperiences([], [tombstonedEntry], false, new Set(['org-deleted'])) + + expect(result.toCreate).toEqual([]) + expect(result.toDelete).toEqual([]) + expect(result.toUpdate.size).toBe(0) + }) + + it('soft-deletes an enrichment-owned row the provider no longer supplies, but keeps protected rows', () => { + const uiRow = oldRow({ id: 'row-ui', orgId: 'org-ui', source: OrganizationSource.UI }) + const droppedRow = oldRow({ id: 'row-dropped', orgId: 'org-dropped' }) + const result = prepareWorkExperiences([uiRow, droppedRow], [], false, new Set()) + + expect(result.toDelete).toEqual([droppedRow]) + expect(result.toCreate).toEqual([]) + expect(result.toUpdate.size).toBe(0) + }) + + it('makes no writes when the payload matches the existing rows exactly', () => { + const existing = oldRow() + const result = prepareWorkExperiences([existing], [newEntry()], false, new Set()) + + expect(result.toCreate).toEqual([]) + expect(result.toDelete).toEqual([]) + expect(result.toUpdate.size).toBe(0) + }) + + it('updates only the changed fields when the payload shifts a date', () => { + const existing = oldRow() + const shifted = newEntry({ endDate: '2021-12-31' }) + const result = prepareWorkExperiences([existing], [shifted], false, new Set()) + + expect(result.toCreate).toEqual([]) + expect(result.toDelete).toEqual([]) + expect(result.toUpdate.get(existing)).toEqual({ dateEnd: '2021-12-31' }) + }) + + it('fills a UI row null dateEnd from a matching provider entry', () => { + const uiRow = oldRow({ id: 'row-ui', source: OrganizationSource.UI, dateEnd: null }) + const matching = newEntry({ endDate: '2021-01-01' }) + const result = prepareWorkExperiences([uiRow], [matching], false, new Set()) + + expect(result.toUpdate.get(uiRow)).toEqual({ dateEnd: '2021-01-01' }) + expect(result.toCreate).toEqual([]) + expect(result.toDelete).toEqual([]) + }) + + it('drops a provider entry that conflicts with a manually-set UI dateEnd instead of applying it', () => { + const uiRow = oldRow({ id: 'row-ui', source: OrganizationSource.UI, dateEnd: '2021-06-01' }) + const conflicting = newEntry({ startDate: '2020-01-01', endDate: '2022-01-01' }) + const result = prepareWorkExperiences([uiRow], [conflicting], false, new Set()) + + expect(result.toCreate).toEqual([]) + expect(result.toDelete).toEqual([]) + expect(result.toUpdate.size).toBe(0) + }) + + describe('when isHighConfidenceSourceSelectedForWorkExperiences is true', () => { + it('still filters out tombstoned organizations', () => { + const tombstonedEntry = newEntry({ organizationId: 'org-deleted' }) + const result = prepareWorkExperiences([], [tombstonedEntry], true, new Set(['org-deleted'])) + + expect(result.toCreate).toEqual([]) + }) + + it('excludes entries that duplicate an existing UI-entered work experience', () => { + const uiRow = oldRow({ id: 'row-ui', source: OrganizationSource.UI }) + const duplicate = newEntry() + const result = prepareWorkExperiences([uiRow], [duplicate], true, new Set()) + + expect(result.toCreate).toEqual([]) + expect(result.toDelete).toEqual([]) + expect(result.toUpdate.size).toBe(0) + }) + + it('creates entries that do not overlap with any UI work experience', () => { + const freshEntry = newEntry({ organizationId: 'org-2', title: 'Manager' }) + const result = prepareWorkExperiences([], [freshEntry], true, new Set()) + + expect(result.toCreate).toEqual([freshEntry]) + }) + }) +}) + +describe('hasMemberOrganizationTimelineChange', () => { + it('returns false when the delete and create sets cover the same organization and dates', () => { + const toDelete = [oldRow({ orgId: 'org-1', dateStart: '2020-01-01', dateEnd: '2021-01-01' })] + const toCreate = [ + newEntry({ + organizationId: 'org-1', + title: 'Different title, same timeline', + startDate: '2020-01-01', + endDate: '2021-01-01', + }), + ] + + expect(hasMemberOrganizationTimelineChange(toDelete, toCreate)).toBe(false) + }) + + it('returns true when the create set introduces a different timeline', () => { + const toDelete = [oldRow({ orgId: 'org-1', dateStart: '2020-01-01', dateEnd: '2021-01-01' })] + const toCreate = [ + newEntry({ organizationId: 'org-1', startDate: '2020-01-01', endDate: '2022-06-01' }), + ] + + expect(hasMemberOrganizationTimelineChange(toDelete, toCreate)).toBe(true) + }) + + it('returns true when the number of affiliations changes', () => { + const toDelete = [oldRow({ orgId: 'org-1', dateStart: '2020-01-01', dateEnd: '2021-01-01' })] + const toCreate = [ + newEntry({ organizationId: 'org-1', startDate: '2020-01-01', endDate: '2021-01-01' }), + newEntry({ organizationId: 'org-2', startDate: '2021-01-02', endDate: null }), + ] + + expect(hasMemberOrganizationTimelineChange(toDelete, toCreate)).toBe(true) + }) +}) diff --git a/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.ts b/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.ts new file mode 100644 index 0000000000..3d81e0c2c2 --- /dev/null +++ b/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.ts @@ -0,0 +1,197 @@ +import { generateUUIDv1 } from '@crowd/common' +import { IMemberOrganizationData, OrganizationSource } from '@crowd/types' + +import { IMemberEnrichmentDataNormalizedOrganization } from '../types' + +export interface IWorkExperienceChanges { + toDelete: IMemberOrganizationData[] + toCreate: IMemberEnrichmentDataNormalizedOrganization[] + // eslint-disable-next-line @typescript-eslint/no-explicit-any + toUpdate: Map> +} + +/** + * Returns true when the set of (orgId, startDate, endDate) tuples differs + * between deletes and creates. Fields like title or source don't affect + * the affiliation timeline, so they're intentionally ignored. + */ +export function hasMemberOrganizationTimelineChange( + toDelete: IMemberOrganizationData[], + toCreate: IMemberEnrichmentDataNormalizedOrganization[], +): boolean { + const toKey = (orgId: string, start: string | null | undefined, end: string | null | undefined) => + `${orgId}|${start ? start.substring(0, 10) : ''}|${end ? end.substring(0, 10) : ''}` + + const deletedKeys = new Set(toDelete.map((d) => toKey(d.orgId, d.dateStart, d.dateEnd))) + const createdKeys = new Set(toCreate.map((c) => toKey(c.organizationId, c.startDate, c.endDate))) + + if (deletedKeys.size !== createdKeys.size) return true + for (const key of deletedKeys) { + if (!createdKeys.has(key)) return true + } + return false +} + +/** + * Reconciles enrichment-owned memberOrganizations rows against the incoming payload + * in place: matched rows are updated (only the fields that actually changed), unmatched + * old rows are deleted, unmatched new entries are created. Never touches UI/project-registry + * or verified rows — callers must exclude those from oldEnrichmentRows. + */ +function reconcileEnrichmentOrgs( + oldEnrichmentRows: IMemberOrganizationData[], + newEntries: IMemberEnrichmentDataNormalizedOrganization[], +): IWorkExperienceChanges { + const normalizeTitle = (title: string | null | undefined) => (title ?? '').trim().toLowerCase() + + const oldByOrg = new Map() + for (const old of oldEnrichmentRows) { + const bucket = oldByOrg.get(old.orgId) ?? [] + bucket.push(old) + oldByOrg.set(old.orgId, bucket) + } + + const toCreate: IMemberEnrichmentDataNormalizedOrganization[] = [] + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const toUpdate: Map> = new Map() + const matchedOldIds = new Set() + + for (const entry of newEntries) { + const candidates = (oldByOrg.get(entry.organizationId) ?? []).filter( + (c) => !matchedOldIds.has(c.id), + ) + const match = + candidates.find((c) => normalizeTitle(c.jobTitle) === normalizeTitle(entry.title)) ?? + candidates[0] + + if (!match) { + toCreate.push(entry) + continue + } + + matchedOldIds.add(match.id) + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const toUpdateInner: Record = {} + if (entry.title !== undefined && entry.title !== match.jobTitle) { + toUpdateInner.title = entry.title + } + if (entry.startDate !== match.dateStart) { + toUpdateInner.dateStart = entry.startDate + } + if (entry.endDate !== match.dateEnd) { + toUpdateInner.dateEnd = entry.endDate + } + if (Object.keys(toUpdateInner).length > 0) { + toUpdate.set(match, toUpdateInner) + } + } + + const toDelete = oldEnrichmentRows.filter((old) => !matchedOldIds.has(old.id)) + + return { toDelete, toCreate, toUpdate } +} + +export function prepareWorkExperiences( + oldVersion: IMemberOrganizationData[], + newVersion: IMemberEnrichmentDataNormalizedOrganization[], + isHighConfidenceSourceSelectedForWorkExperiences: boolean, + deletedOrganizationIds: Set, +): IWorkExperienceChanges { + // UI and project-registry rows are manual input; a verified row is a human decision too — + // the worker never deletes or updates any of them. + const oldEnrichmentRows = oldVersion.filter( + (c) => + c.source !== OrganizationSource.UI && + c.source !== OrganizationSource.PROJECT_REGISTRY && + c.verified !== true, + ) + + // never recreate an affiliation that a person deleted on purpose — providers keep resupplying it + newVersion = newVersion.filter((e) => !deletedOrganizationIds.has(e.organizationId)) + + if (isHighConfidenceSourceSelectedForWorkExperiences) { + const uiEntries = oldVersion.filter((c) => c.source === OrganizationSource.UI) + const filteredNewVersion = newVersion.filter( + (e) => + !uiEntries.some( + (ui) => + e.title === ui.jobTitle && + e.identities && + e.identities.some((i) => i.organizationId === ui.orgId), + ), + ) + return reconcileEnrichmentOrgs(oldEnrichmentRows, filteredNewVersion) + } + + // sort both versions by start date and only use manual changes from the current version + const orderedCurrentVersion = oldVersion + .filter((c) => c.source === OrganizationSource.UI) + .sort((a, b) => { + // If either value is null/undefined, move it to the beginning + if (!a.dateStart && !b.dateStart) return 0 + if (!a.dateStart) return -1 + if (!b.dateStart) return 1 + + // Compare dates if both values exist + return new Date(a.dateStart as string).getTime() - new Date(b.dateStart as string).getTime() + }) + + let orderedNewVersion = newVersion.sort((a, b) => { + // If either value is null/undefined, move it to the beginning + if (!a.startDate && !b.startDate) return 0 + if (!a.startDate) return -1 + if (!b.startDate) return 1 + + // Compare dates if both values exist + return new Date(a.startDate as string).getTime() - new Date(b.startDate as string).getTime() + }) + + // set ids and new flag to new versions just so we can easily manipulate the array later + for (const exp of orderedNewVersion) { + exp.id = generateUUIDv1() + } + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const uiDateEndFills: Map> = new Map() + + // we iterate through the existing version experiences to see if update is needed + for (const current of orderedCurrentVersion) { + // try and find a matching experience in the new versions by title + const match = orderedNewVersion.find( + (e) => + e.title === current.jobTitle && + e.identities && + e.identities.some((e) => e.organizationId === current.orgId), + ) + + // if we found a match we can check if we need something to update + if ( + match && + current.dateStart === match.startDate && + current.dateEnd === null && + match.endDate !== null + ) { + uiDateEndFills.set(current, { dateEnd: match.endDate }) + + // remove the match from the new version array so we later don't process it again + orderedNewVersion = orderedNewVersion.filter((e) => e.id !== match.id) + } else if ( + match && + (current.dateStart !== match.startDate || current.dateEnd !== null || match.endDate === null) + ) { + // there's an incoming work experiences, but it's conflicting with the existing manually updated data + // we shouldn't add or update anything when this happens + // we can only update dateEnd of existing manually changed data, when it has a null dateEnd + orderedNewVersion = orderedNewVersion.filter((e) => e.id !== match.id) + } + // if we didn't find a match we should just leave it as it is in the database since it was manual input + } + + const results = reconcileEnrichmentOrgs(oldEnrichmentRows, orderedNewVersion) + for (const [current, toUpdateInner] of uiDateEndFills) { + results.toUpdate.set(current, toUpdateInner) + } + + return results +} diff --git a/services/apps/members_enrichment_worker/vitest.config.ts b/services/apps/members_enrichment_worker/vitest.config.ts new file mode 100644 index 0000000000..530cc66032 --- /dev/null +++ b/services/apps/members_enrichment_worker/vitest.config.ts @@ -0,0 +1,13 @@ +import { defineConfig } from 'vitest/config' + +export default defineConfig({ + test: { + environment: 'node', + include: ['src/**/*.test.ts'], + server: { + deps: { + inline: [/@crowd\//], + }, + }, + }, +}) diff --git a/services/libs/data-access-layer/src/members/organizations.ts b/services/libs/data-access-layer/src/members/organizations.ts index 3a31a2a9d0..e6ebd7863a 100644 --- a/services/libs/data-access-layer/src/members/organizations.ts +++ b/services/libs/data-access-layer/src/members/organizations.ts @@ -513,15 +513,20 @@ export async function deleteMemberOrganizations( memberId: string, ids?: string[], softDelete = true, + deletedBy?: string, ): Promise { - // Base query depends on soft vs hard delete + // Base query depends on soft vs hard delete. deletedBy marks this as a human-initiated + // delete — the enrichment worker's own rebuild deletes must never set it, since only a + // human delete is meant to permanently block the affiliation from being recreated. const baseQuery = softDelete - ? 'UPDATE "memberOrganizations" SET "deletedAt" = NOW()' + ? deletedBy + ? 'UPDATE "memberOrganizations" SET "deletedAt" = NOW(), "deletedBy" = $(deletedBy)' + : 'UPDATE "memberOrganizations" SET "deletedAt" = NOW()' : 'DELETE FROM "memberOrganizations"' // Build WHERE clause const conditions = ['"memberId" = $(memberId)'] - const params: Record = { memberId } + const params: Record = { memberId, deletedBy } if (ids?.length) { conditions.push(`"id" IN ($(ids:csv))`) diff --git a/services/libs/data-access-layer/src/old/apps/members_enrichment_worker/index.ts b/services/libs/data-access-layer/src/old/apps/members_enrichment_worker/index.ts index e6421d433e..61edea5ed6 100644 --- a/services/libs/data-access-layer/src/old/apps/members_enrichment_worker/index.ts +++ b/services/libs/data-access-layer/src/old/apps/members_enrichment_worker/index.ts @@ -44,6 +44,8 @@ export async function fetchMemberDataForLLMSquashing( mo."dateStart", mo."dateEnd", mo.source, + mo.verified, + mo."verifiedBy", jsonb_agg(jsonb_build_object( 'organizationId', oi."organizationId", 'platform', oi.platform, @@ -63,7 +65,7 @@ export async function fetchMemberDataForLLMSquashing( from "memberOrganizations" mo where mo."memberId" = $(memberId) and mo."deletedAt" is not null - and mo.source not in ('ui', 'project-registry')) + and mo."deletedBy" is not null) select m."displayName", m.attributes, m."manuallyChangedFields", @@ -90,6 +92,8 @@ export async function fetchMemberDataForLLMSquashing( mo."dateStart", mo."dateEnd", mo.source, + mo.verified, + mo."verifiedBy", coalesce(mo.identities, '[]'::jsonb) as identities) r) ) from member_orgs mo @@ -571,7 +575,7 @@ export async function updateMemberOrg( return null } - const sets = keys.map((k) => `"${k}" = $(${k})`) + const sets = [...keys.map((k) => `"${k}" = $(${k})`), `"updatedAt" = now()`] const result = await tx.oneOrNone( ` diff --git a/services/libs/types/src/enrichment.ts b/services/libs/types/src/enrichment.ts index e1b21d95a4..e2e72e8f7a 100644 --- a/services/libs/types/src/enrichment.ts +++ b/services/libs/types/src/enrichment.ts @@ -40,6 +40,8 @@ export interface IMemberOrganizationData { dateStart: string dateEnd: string source: string + verified?: boolean + verifiedBy?: string | null identities?: IOrganizationIdentity[] } @@ -59,7 +61,7 @@ export interface IMemberOriginalData { // memberOrganizations table data organizations: IMemberOrganizationData[] - // memberOrganizations rows manually deleted, source not UI/PROJECT_REGISTRY — tombstones enrichment must not recreate + // memberOrganizations rows manually deleted by a person (deletedBy set) deletedOrganizations: IDeletedMemberOrganizationData[] } From 236bc7a2eebfe245b2c711871c2c4e7a246e71d9 Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Thu, 13 Aug 2026 19:38:40 +0200 Subject: [PATCH 3/8] fix: update-model store deletedBy Signed-off-by: Umberto Sgueglia --- .../workExperienceReconciliation.test.ts | 34 +++++++++++++++++++ .../workExperienceReconciliation.ts | 30 ++++++++++++---- .../src/members/organizations.ts | 5 ++- 3 files changed, 60 insertions(+), 9 deletions(-) diff --git a/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.test.ts b/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.test.ts index b0664688d7..08bcef2a9a 100644 --- a/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.test.ts +++ b/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.test.ts @@ -65,6 +65,22 @@ describe('prepareWorkExperiences', () => { expect(result.toUpdate.has(verified)).toBe(false) }) + it('does not insert a duplicate affiliation for a provider entry matching a verified org+title', () => { + const verified = oldRow({ id: 'row-verified', verified: true, verifiedBy: 'jane' }) + const duplicate = newEntry({ startDate: '2021-06-01', endDate: '2022-01-01' }) + const result = prepareWorkExperiences([verified], [duplicate], false, new Set()) + + expect(result.toCreate).toEqual([]) + }) + + it('still creates an entry for a distinct role at the same org as a verified row', () => { + const verified = oldRow({ id: 'row-verified', verified: true, verifiedBy: 'jane' }) + const distinctRole = newEntry({ title: 'Manager', startDate: '2022-01-01', endDate: null }) + const result = prepareWorkExperiences([verified], [distinctRole], false, new Set()) + + expect(result.toCreate).toEqual([distinctRole]) + }) + it('never recreates an organization a person deleted on purpose (tombstoned)', () => { const tombstonedEntry = newEntry({ organizationId: 'org-deleted' }) const result = prepareWorkExperiences([], [tombstonedEntry], false, new Set(['org-deleted'])) @@ -103,6 +119,24 @@ describe('prepareWorkExperiences', () => { expect(result.toUpdate.get(existing)).toEqual({ dateEnd: '2021-12-31' }) }) + it('adopts the incoming source on a matched row when the provider source changed', () => { + const existing = oldRow({ source: OrganizationSource.ENRICHMENT_PROGAI }) + const resupplied = newEntry({ source: OrganizationSource.ENRICHMENT_CRUSTDATA }) + const result = prepareWorkExperiences([existing], [resupplied], false, new Set()) + + expect(result.toUpdate.get(existing)).toEqual({ + source: OrganizationSource.ENRICHMENT_CRUSTDATA, + }) + }) + + it('does not treat a date-only vs full-timestamp difference as a change', () => { + const existing = oldRow({ dateStart: '2020-01-01T00:00:00.000Z', dateEnd: null }) + const resupplied = newEntry({ startDate: '2020-01-01', endDate: null }) + const result = prepareWorkExperiences([existing], [resupplied], false, new Set()) + + expect(result.toUpdate.size).toBe(0) + }) + it('fills a UI row null dateEnd from a matching provider entry', () => { const uiRow = oldRow({ id: 'row-ui', source: OrganizationSource.UI, dateEnd: null }) const matching = newEntry({ endDate: '2021-01-01' }) diff --git a/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.ts b/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.ts index 3d81e0c2c2..1ab05bc6ce 100644 --- a/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.ts +++ b/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.ts @@ -10,6 +10,9 @@ export interface IWorkExperienceChanges { toUpdate: Map> } +const normalizeTitle = (title: string | null | undefined) => (title ?? '').trim().toLowerCase() +const normalizeDate = (date: string | null | undefined) => (date ? date.substring(0, 10) : '') + /** * Returns true when the set of (orgId, startDate, endDate) tuples differs * between deletes and creates. Fields like title or source don't affect @@ -20,7 +23,7 @@ export function hasMemberOrganizationTimelineChange( toCreate: IMemberEnrichmentDataNormalizedOrganization[], ): boolean { const toKey = (orgId: string, start: string | null | undefined, end: string | null | undefined) => - `${orgId}|${start ? start.substring(0, 10) : ''}|${end ? end.substring(0, 10) : ''}` + `${orgId}|${normalizeDate(start)}|${normalizeDate(end)}` const deletedKeys = new Set(toDelete.map((d) => toKey(d.orgId, d.dateStart, d.dateEnd))) const createdKeys = new Set(toCreate.map((c) => toKey(c.organizationId, c.startDate, c.endDate))) @@ -36,14 +39,13 @@ export function hasMemberOrganizationTimelineChange( * Reconciles enrichment-owned memberOrganizations rows against the incoming payload * in place: matched rows are updated (only the fields that actually changed), unmatched * old rows are deleted, unmatched new entries are created. Never touches UI/project-registry - * or verified rows — callers must exclude those from oldEnrichmentRows. + * or verified rows — callers must exclude those from oldEnrichmentRows and filter matching + * newEntries out beforehand. */ function reconcileEnrichmentOrgs( oldEnrichmentRows: IMemberOrganizationData[], newEntries: IMemberEnrichmentDataNormalizedOrganization[], ): IWorkExperienceChanges { - const normalizeTitle = (title: string | null | undefined) => (title ?? '').trim().toLowerCase() - const oldByOrg = new Map() for (const old of oldEnrichmentRows) { const bucket = oldByOrg.get(old.orgId) ?? [] @@ -76,12 +78,15 @@ function reconcileEnrichmentOrgs( if (entry.title !== undefined && entry.title !== match.jobTitle) { toUpdateInner.title = entry.title } - if (entry.startDate !== match.dateStart) { + if (normalizeDate(entry.startDate) !== normalizeDate(match.dateStart)) { toUpdateInner.dateStart = entry.startDate } - if (entry.endDate !== match.dateEnd) { + if (normalizeDate(entry.endDate) !== normalizeDate(match.dateEnd)) { toUpdateInner.dateEnd = entry.endDate } + if (entry.source !== undefined && entry.source !== match.source) { + toUpdateInner.source = entry.source + } if (Object.keys(toUpdateInner).length > 0) { toUpdate.set(match, toUpdateInner) } @@ -110,6 +115,19 @@ export function prepareWorkExperiences( // never recreate an affiliation that a person deleted on purpose — providers keep resupplying it newVersion = newVersion.filter((e) => !deletedOrganizationIds.has(e.organizationId)) + // verified rows are excluded from oldEnrichmentRows above, so a matching provider entry + // must be dropped here too, or it lands in toCreate as a conflicting duplicate + const verifiedRows = oldVersion.filter((c) => c.verified === true) + newVersion = newVersion.filter( + (e) => + !verifiedRows.some( + (v) => + normalizeTitle(v.jobTitle) === normalizeTitle(e.title) && + e.identities && + e.identities.some((i) => i.organizationId === v.orgId), + ), + ) + if (isHighConfidenceSourceSelectedForWorkExperiences) { const uiEntries = oldVersion.filter((c) => c.source === OrganizationSource.UI) const filteredNewVersion = newVersion.filter( diff --git a/services/libs/data-access-layer/src/members/organizations.ts b/services/libs/data-access-layer/src/members/organizations.ts index e6ebd7863a..b907f47339 100644 --- a/services/libs/data-access-layer/src/members/organizations.ts +++ b/services/libs/data-access-layer/src/members/organizations.ts @@ -515,9 +515,8 @@ export async function deleteMemberOrganizations( softDelete = true, deletedBy?: string, ): Promise { - // Base query depends on soft vs hard delete. deletedBy marks this as a human-initiated - // delete — the enrichment worker's own rebuild deletes must never set it, since only a - // human delete is meant to permanently block the affiliation from being recreated. + // deletedBy marks a human delete; the enrichment worker's own rebuild deletes must + // never set it, since only a human delete permanently blocks recreation. const baseQuery = softDelete ? deletedBy ? 'UPDATE "memberOrganizations" SET "deletedAt" = NOW(), "deletedBy" = $(deletedBy)' From 71e984bb688cceb4d2312bd47c6d2f9551572ebd Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Thu, 13 Aug 2026 19:51:31 +0200 Subject: [PATCH 4/8] fix: update-model store deletedBy Signed-off-by: Umberto Sgueglia --- .../members_enrichment_worker/src/activities/enrichment.ts | 2 +- services/libs/types/src/enrichment.ts | 5 +++-- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/services/apps/members_enrichment_worker/src/activities/enrichment.ts b/services/apps/members_enrichment_worker/src/activities/enrichment.ts index 4b03e23611..36b7eb6c13 100644 --- a/services/apps/members_enrichment_worker/src/activities/enrichment.ts +++ b/services/apps/members_enrichment_worker/src/activities/enrichment.ts @@ -604,7 +604,7 @@ export async function updateMemberUsingSquashedPayload( existingMemberData.organizations, squashedPayload.memberOrganizations, isHighConfidenceSourceSelectedForWorkExperiences, - new Set(existingMemberData.deletedOrganizations.map((o) => o.orgId)), + new Set((existingMemberData.deletedOrganizations ?? []).map((o) => o.orgId)), ) // Skip the refresh when the timeline that drives activityRelations hasn't changed — diff --git a/services/libs/types/src/enrichment.ts b/services/libs/types/src/enrichment.ts index e2e72e8f7a..4d0581e674 100644 --- a/services/libs/types/src/enrichment.ts +++ b/services/libs/types/src/enrichment.ts @@ -61,8 +61,9 @@ export interface IMemberOriginalData { // memberOrganizations table data organizations: IMemberOrganizationData[] - // memberOrganizations rows manually deleted by a person (deletedBy set) - deletedOrganizations: IDeletedMemberOrganizationData[] + // memberOrganizations rows manually deleted by a person (deletedBy set). Optional because + // in-flight Temporal histories may carry a pre-rollout result that predates this field. + deletedOrganizations?: IDeletedMemberOrganizationData[] } export interface IOrganizationEnrichmentCache { From 7bbf97e542a547b1178b8d5aa38181e8cc4e4207 Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Thu, 13 Aug 2026 20:39:52 +0200 Subject: [PATCH 5/8] fix: update-model store deletedBy Signed-off-by: Umberto Sgueglia --- .../workExperienceReconciliation.test.ts | 17 +++++++++++++++++ .../activities/workExperienceReconciliation.ts | 6 ++++++ 2 files changed, 23 insertions(+) diff --git a/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.test.ts b/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.test.ts index 08bcef2a9a..f68ffbb4ed 100644 --- a/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.test.ts +++ b/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.test.ts @@ -137,6 +137,23 @@ describe('prepareWorkExperiences', () => { expect(result.toUpdate.size).toBe(0) }) + it('matches multiple same-title stints at one org by date, regardless of payload order', () => { + const stintA = oldRow({ id: 'row-a', dateStart: '2018-01-01', dateEnd: '2019-01-01' }) + const stintB = oldRow({ id: 'row-b', dateStart: '2020-01-01', dateEnd: '2021-01-01' }) + const entryForB = newEntry({ startDate: '2020-01-01', endDate: '2021-01-01' }) + const entryForA = newEntry({ startDate: '2018-01-01', endDate: '2019-01-01' }) + const result = prepareWorkExperiences( + [stintA, stintB], + [entryForB, entryForA], + false, + new Set(), + ) + + expect(result.toDelete).toEqual([]) + expect(result.toCreate).toEqual([]) + expect(result.toUpdate.size).toBe(0) + }) + it('fills a UI row null dateEnd from a matching provider entry', () => { const uiRow = oldRow({ id: 'row-ui', source: OrganizationSource.UI, dateEnd: null }) const matching = newEntry({ endDate: '2021-01-01' }) diff --git a/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.ts b/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.ts index 1ab05bc6ce..a19241822b 100644 --- a/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.ts +++ b/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.ts @@ -63,6 +63,12 @@ function reconcileEnrichmentOrgs( (c) => !matchedOldIds.has(c.id), ) const match = + candidates.find( + (c) => + normalizeTitle(c.jobTitle) === normalizeTitle(entry.title) && + normalizeDate(c.dateStart) === normalizeDate(entry.startDate) && + normalizeDate(c.dateEnd) === normalizeDate(entry.endDate), + ) ?? candidates.find((c) => normalizeTitle(c.jobTitle) === normalizeTitle(entry.title)) ?? candidates[0] From 29e222a82c37ed638c323e0f0247ad17d5d35a25 Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Fri, 14 Aug 2026 09:10:27 +0200 Subject: [PATCH 6/8] fix: update-model store deletedBy Signed-off-by: Umberto Sgueglia --- .../workExperienceReconciliation.test.ts | 84 +++++++++++++++++++ .../workExperienceReconciliation.ts | 84 +++++++++++++++++-- 2 files changed, 160 insertions(+), 8 deletions(-) diff --git a/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.test.ts b/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.test.ts index f68ffbb4ed..ea42a9eb52 100644 --- a/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.test.ts +++ b/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.test.ts @@ -154,6 +154,90 @@ describe('prepareWorkExperiences', () => { expect(result.toUpdate.size).toBe(0) }) + it('survives two active stints swapping their date ranges instead of dropping one', () => { + // distinct titles force title-based matching, so the swap can't be masked by + // exact-date-match-first picking up the "other" row by its still-current dates + const stintEngineer = oldRow({ + id: 'row-engineer', + jobTitle: 'Engineer', + dateStart: '2018-01-01', + dateEnd: '2019-01-01', + }) + const stintManager = oldRow({ + id: 'row-manager', + jobTitle: 'Manager', + dateStart: '2020-01-01', + dateEnd: '2021-01-01', + }) + // provider now reports each role with the other role's current date range + const entryForEngineer = newEntry({ + title: 'Engineer', + startDate: '2020-01-01', + endDate: '2021-01-01', + }) + const entryForManager = newEntry({ + title: 'Manager', + startDate: '2018-01-01', + endDate: '2019-01-01', + }) + + const result = prepareWorkExperiences( + [stintEngineer, stintManager], + [entryForEngineer, entryForManager], + false, + new Set(), + ) + + // a genuine cycle (each target tuple is held by the other row) can't be resolved by + // reordering, so both stints fall back to delete+create pairs instead of an in-place update + expect(result.toDelete.length).toBe(2) + expect(result.toCreate.length).toBe(2) + expect(result.toUpdate.size).toBe(0) + + const finalTuplesByTitle = new Map() + for (const [oldOrg, changes] of result.toUpdate) { + finalTuplesByTitle.set( + changes.title ?? oldOrg.jobTitle, + `${changes.dateStart ?? oldOrg.dateStart}|${changes.dateEnd ?? oldOrg.dateEnd}`, + ) + } + for (const created of result.toCreate) { + finalTuplesByTitle.set(created.title, `${created.startDate}|${created.endDate}`) + } + expect(finalTuplesByTitle.get('Engineer')).toBe('2020-01-01|2021-01-01') + expect(finalTuplesByTitle.get('Manager')).toBe('2018-01-01|2019-01-01') + }) + + it('resolves a date-shift chain via in-place updates without falling back to delete+create', () => { + // row-x's new target tuple is free from the start, so it can update first and free + // its own old tuple for row-y — a chain, not a cycle, so no fallback is needed + const rowX = oldRow({ + id: 'row-x', + jobTitle: 'Engineer', + dateStart: '2018-01-01', + dateEnd: '2019-01-01', + }) + const rowY = oldRow({ + id: 'row-y', + jobTitle: 'Manager', + dateStart: '2019-01-01', + dateEnd: '2020-01-01', + }) + const entryForX = newEntry({ + title: 'Engineer', + startDate: '2019-01-01', + endDate: '2020-01-01', + }) + const entryForY = newEntry({ title: 'Manager', startDate: '2020-01-01', endDate: '2021-01-01' }) + + const result = prepareWorkExperiences([rowX, rowY], [entryForX, entryForY], false, new Set()) + + expect(result.toDelete).toEqual([]) + expect(result.toCreate).toEqual([]) + expect(result.toUpdate.get(rowX)).toEqual({ dateStart: '2019-01-01', dateEnd: '2020-01-01' }) + expect(result.toUpdate.get(rowY)).toEqual({ dateStart: '2020-01-01', dateEnd: '2021-01-01' }) + }) + it('fills a UI row null dateEnd from a matching provider entry', () => { const uiRow = oldRow({ id: 'row-ui', source: OrganizationSource.UI, dateEnd: null }) const matching = newEntry({ endDate: '2021-01-01' }) diff --git a/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.ts b/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.ts index a19241822b..ca5d66f2bd 100644 --- a/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.ts +++ b/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.ts @@ -12,6 +12,11 @@ export interface IWorkExperienceChanges { const normalizeTitle = (title: string | null | undefined) => (title ?? '').trim().toLowerCase() const normalizeDate = (date: string | null | undefined) => (date ? date.substring(0, 10) : '') +const dateTupleKey = ( + orgId: string, + start: string | null | undefined, + end: string | null | undefined, +) => `${orgId}|${normalizeDate(start)}|${normalizeDate(end)}` /** * Returns true when the set of (orgId, startDate, endDate) tuples differs @@ -22,11 +27,10 @@ export function hasMemberOrganizationTimelineChange( toDelete: IMemberOrganizationData[], toCreate: IMemberEnrichmentDataNormalizedOrganization[], ): boolean { - const toKey = (orgId: string, start: string | null | undefined, end: string | null | undefined) => - `${orgId}|${normalizeDate(start)}|${normalizeDate(end)}` - - const deletedKeys = new Set(toDelete.map((d) => toKey(d.orgId, d.dateStart, d.dateEnd))) - const createdKeys = new Set(toCreate.map((c) => toKey(c.organizationId, c.startDate, c.endDate))) + const deletedKeys = new Set(toDelete.map((d) => dateTupleKey(d.orgId, d.dateStart, d.dateEnd))) + const createdKeys = new Set( + toCreate.map((c) => dateTupleKey(c.organizationId, c.startDate, c.endDate)), + ) if (deletedKeys.size !== createdKeys.size) return true for (const key of deletedKeys) { @@ -35,6 +39,60 @@ export function hasMemberOrganizationTimelineChange( return false } +interface IPendingOrgUpdate { + oldRow: IMemberOrganizationData + entry: IMemberEnrichmentDataNormalizedOrganization + // eslint-disable-next-line @typescript-eslint/no-explicit-any + toUpdateInner: Record +} + +/** + * The unique index on (memberId, organizationId, dateStart, dateEnd) means an in-place update + * can only be applied while its target tuple isn't still held by another row. Schedules + * date-changing updates in an order where every target tuple is free by the time it runs; + * rows caught in a genuine swap/cycle (no valid order exists) fall back to delete+create, + * which the caller always applies before any update. + */ +function scheduleDateChangingUpdates( + pending: IPendingOrgUpdate[], + // eslint-disable-next-line @typescript-eslint/no-explicit-any + toUpdate: Map>, + toCreate: IMemberEnrichmentDataNormalizedOrganization[], + toDelete: IMemberOrganizationData[], +) { + const targetKey = (u: IPendingOrgUpdate) => + dateTupleKey( + u.oldRow.orgId, + u.toUpdateInner.dateStart ?? u.oldRow.dateStart, + u.toUpdateInner.dateEnd ?? u.oldRow.dateEnd, + ) + const currentKey = (u: IPendingOrgUpdate) => + dateTupleKey(u.oldRow.orgId, u.oldRow.dateStart, u.oldRow.dateEnd) + + let remaining = pending + let progress = true + while (remaining.length > 0 && progress) { + progress = false + const stillHeldKeys = new Set(remaining.map(currentKey)) + const next: IPendingOrgUpdate[] = [] + for (const u of remaining) { + if (!stillHeldKeys.has(targetKey(u))) { + toUpdate.set(u.oldRow, u.toUpdateInner) + progress = true + } else { + next.push(u) + } + } + remaining = next + } + + // a genuine cycle — no sequential order frees every target tuple in time + for (const u of remaining) { + toDelete.push(u.oldRow) + toCreate.push(u.entry) + } +} + /** * Reconciles enrichment-owned memberOrganizations rows against the incoming payload * in place: matched rows are updated (only the fields that actually changed), unmatched @@ -57,6 +115,7 @@ function reconcileEnrichmentOrgs( // eslint-disable-next-line @typescript-eslint/no-explicit-any const toUpdate: Map> = new Map() const matchedOldIds = new Set() + const pendingDateChanges: IPendingOrgUpdate[] = [] for (const entry of newEntries) { const candidates = (oldByOrg.get(entry.organizationId) ?? []).filter( @@ -84,22 +143,31 @@ function reconcileEnrichmentOrgs( if (entry.title !== undefined && entry.title !== match.jobTitle) { toUpdateInner.title = entry.title } - if (normalizeDate(entry.startDate) !== normalizeDate(match.dateStart)) { + const startChanged = normalizeDate(entry.startDate) !== normalizeDate(match.dateStart) + const endChanged = normalizeDate(entry.endDate) !== normalizeDate(match.dateEnd) + if (startChanged) { toUpdateInner.dateStart = entry.startDate } - if (normalizeDate(entry.endDate) !== normalizeDate(match.dateEnd)) { + if (endChanged) { toUpdateInner.dateEnd = entry.endDate } if (entry.source !== undefined && entry.source !== match.source) { toUpdateInner.source = entry.source } - if (Object.keys(toUpdateInner).length > 0) { + if (Object.keys(toUpdateInner).length === 0) { + continue + } + if (startChanged || endChanged) { + pendingDateChanges.push({ oldRow: match, entry, toUpdateInner }) + } else { toUpdate.set(match, toUpdateInner) } } const toDelete = oldEnrichmentRows.filter((old) => !matchedOldIds.has(old.id)) + scheduleDateChangingUpdates(pendingDateChanges, toUpdate, toCreate, toDelete) + return { toDelete, toCreate, toUpdate } } From e2e41b928f3df61898d67c603150d727e5fc598d Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Fri, 14 Aug 2026 11:10:42 +0200 Subject: [PATCH 7/8] fix: fix logic Signed-off-by: Umberto Sgueglia --- .../members_enrichment_worker/package.json | 6 ++---- .../workExperienceReconciliation.test.ts | 3 ++- .../workExperienceReconciliation.ts | 21 ++++++------------- .../vitest.config.ts | 13 ------------ 4 files changed, 10 insertions(+), 33 deletions(-) delete mode 100644 services/apps/members_enrichment_worker/vitest.config.ts diff --git a/services/apps/members_enrichment_worker/package.json b/services/apps/members_enrichment_worker/package.json index d1446827fc..64da9ef784 100644 --- a/services/apps/members_enrichment_worker/package.json +++ b/services/apps/members_enrichment_worker/package.json @@ -10,8 +10,7 @@ "lint": "npx eslint --ext .ts src --max-warnings=0", "format": "npx prettier --write \"src/**/*.ts\"", "format-check": "npx prettier --check .", - "tsc-check": "tsc --noEmit", - "test": "vitest run" + "tsc-check": "tsc --noEmit" }, "dependencies": { "@crowd/archetype-standard": "workspace:*", @@ -36,7 +35,6 @@ "devDependencies": { "@types/node": "^20.8.2", "@types/uuid": "~9.0.6", - "nodemon": "^3.0.1", - "vitest": "^3.2.4" + "nodemon": "^3.0.1" } } diff --git a/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.test.ts b/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.test.ts index ea42a9eb52..3b8945d3ae 100644 --- a/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.test.ts +++ b/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.test.ts @@ -34,9 +34,10 @@ function newEntry( startDate: '2020-01-01', endDate: null, source: OrganizationSource.ENRICHMENT_PROGAI, + // on real data the resolved org lands on organizationId, not on identities — enrichment + // does not backfill identities with the org it just resolved identities: [ { - organizationId, platform: 'linkedin', value: 'org-one', type: OrganizationIdentityType.USERNAME, diff --git a/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.ts b/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.ts index ca5d66f2bd..8820d18f5f 100644 --- a/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.ts +++ b/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.ts @@ -190,28 +190,22 @@ export function prepareWorkExperiences( newVersion = newVersion.filter((e) => !deletedOrganizationIds.has(e.organizationId)) // verified rows are excluded from oldEnrichmentRows above, so a matching provider entry - // must be dropped here too, or it lands in toCreate as a conflicting duplicate + // must be dropped here too, or it lands in toCreate as a conflicting duplicate. Match on + // e.organizationId, not e.identities — enrichment resolves the org onto organizationId + // without necessarily adding it to identities. const verifiedRows = oldVersion.filter((c) => c.verified === true) newVersion = newVersion.filter( (e) => !verifiedRows.some( (v) => - normalizeTitle(v.jobTitle) === normalizeTitle(e.title) && - e.identities && - e.identities.some((i) => i.organizationId === v.orgId), + normalizeTitle(v.jobTitle) === normalizeTitle(e.title) && e.organizationId === v.orgId, ), ) if (isHighConfidenceSourceSelectedForWorkExperiences) { const uiEntries = oldVersion.filter((c) => c.source === OrganizationSource.UI) const filteredNewVersion = newVersion.filter( - (e) => - !uiEntries.some( - (ui) => - e.title === ui.jobTitle && - e.identities && - e.identities.some((i) => i.organizationId === ui.orgId), - ), + (e) => !uiEntries.some((ui) => e.title === ui.jobTitle && e.organizationId === ui.orgId), ) return reconcileEnrichmentOrgs(oldEnrichmentRows, filteredNewVersion) } @@ -251,10 +245,7 @@ export function prepareWorkExperiences( for (const current of orderedCurrentVersion) { // try and find a matching experience in the new versions by title const match = orderedNewVersion.find( - (e) => - e.title === current.jobTitle && - e.identities && - e.identities.some((e) => e.organizationId === current.orgId), + (e) => e.title === current.jobTitle && e.organizationId === current.orgId, ) // if we found a match we can check if we need something to update diff --git a/services/apps/members_enrichment_worker/vitest.config.ts b/services/apps/members_enrichment_worker/vitest.config.ts deleted file mode 100644 index 530cc66032..0000000000 --- a/services/apps/members_enrichment_worker/vitest.config.ts +++ /dev/null @@ -1,13 +0,0 @@ -import { defineConfig } from 'vitest/config' - -export default defineConfig({ - test: { - environment: 'node', - include: ['src/**/*.test.ts'], - server: { - deps: { - inline: [/@crowd\//], - }, - }, - }, -}) From 33d94c44a291d51e9fd15d0ffc2d281b7c102ef9 Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Fri, 14 Aug 2026 11:13:02 +0200 Subject: [PATCH 8/8] fix: update lock Signed-off-by: Umberto Sgueglia --- pnpm-lock.yaml | 23 ++++++++++------------- 1 file changed, 10 insertions(+), 13 deletions(-) diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index dfe2aaf8db..d496467b1a 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1071,9 +1071,6 @@ importers: nodemon: specifier: ^3.0.1 version: 3.1.0 - vitest: - specifier: ^3.2.4 - version: 3.2.4(@types/debug@4.1.12)(@types/node@20.12.7)(jiti@2.4.2)(terser@5.43.1)(tsx@4.7.3)(yaml@2.7.0) services/apps/merge_suggestions_worker: dependencies: @@ -10972,8 +10969,8 @@ snapshots: dependencies: '@aws-crypto/sha256-browser': 3.0.0 '@aws-crypto/sha256-js': 3.0.0 - '@aws-sdk/client-sso-oidc': 3.572.0(@aws-sdk/client-sts@3.572.0) - '@aws-sdk/client-sts': 3.572.0 + '@aws-sdk/client-sso-oidc': 3.572.0 + '@aws-sdk/client-sts': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) '@aws-sdk/core': 3.572.0 '@aws-sdk/credential-provider-node': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0)(@aws-sdk/client-sts@3.572.0) '@aws-sdk/middleware-host-header': 3.567.0 @@ -11167,11 +11164,11 @@ snapshots: transitivePeerDependencies: - aws-crt - '@aws-sdk/client-sso-oidc@3.572.0(@aws-sdk/client-sts@3.572.0)': + '@aws-sdk/client-sso-oidc@3.572.0': dependencies: '@aws-crypto/sha256-browser': 3.0.0 '@aws-crypto/sha256-js': 3.0.0 - '@aws-sdk/client-sts': 3.572.0 + '@aws-sdk/client-sts': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) '@aws-sdk/core': 3.572.0 '@aws-sdk/credential-provider-node': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0)(@aws-sdk/client-sts@3.572.0) '@aws-sdk/middleware-host-header': 3.567.0 @@ -11210,7 +11207,6 @@ snapshots: '@smithy/util-utf8': 2.3.0 tslib: 2.6.2 transitivePeerDependencies: - - '@aws-sdk/client-sts' - aws-crt '@aws-sdk/client-sso@3.556.0': @@ -11386,11 +11382,11 @@ snapshots: transitivePeerDependencies: - aws-crt - '@aws-sdk/client-sts@3.572.0': + '@aws-sdk/client-sts@3.572.0(@aws-sdk/client-sso-oidc@3.572.0)': dependencies: '@aws-crypto/sha256-browser': 3.0.0 '@aws-crypto/sha256-js': 3.0.0 - '@aws-sdk/client-sso-oidc': 3.572.0(@aws-sdk/client-sts@3.572.0) + '@aws-sdk/client-sso-oidc': 3.572.0 '@aws-sdk/core': 3.572.0 '@aws-sdk/credential-provider-node': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0)(@aws-sdk/client-sts@3.572.0) '@aws-sdk/middleware-host-header': 3.567.0 @@ -11429,6 +11425,7 @@ snapshots: '@smithy/util-utf8': 2.3.0 tslib: 2.6.2 transitivePeerDependencies: + - '@aws-sdk/client-sso-oidc' - aws-crt '@aws-sdk/client-sts@3.985.0': @@ -11594,7 +11591,7 @@ snapshots: '@aws-sdk/credential-provider-ini@3.572.0(@aws-sdk/client-sso-oidc@3.572.0)(@aws-sdk/client-sts@3.572.0)': dependencies: - '@aws-sdk/client-sts': 3.572.0 + '@aws-sdk/client-sts': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) '@aws-sdk/credential-provider-env': 3.568.0 '@aws-sdk/credential-provider-process': 3.572.0 '@aws-sdk/credential-provider-sso': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) @@ -11771,7 +11768,7 @@ snapshots: '@aws-sdk/credential-provider-web-identity@3.568.0(@aws-sdk/client-sts@3.572.0)': dependencies: - '@aws-sdk/client-sts': 3.572.0 + '@aws-sdk/client-sts': 3.572.0(@aws-sdk/client-sso-oidc@3.572.0) '@aws-sdk/types': 3.567.0 '@smithy/property-provider': 2.2.0 '@smithy/types': 2.12.0 @@ -12083,7 +12080,7 @@ snapshots: '@aws-sdk/token-providers@3.572.0(@aws-sdk/client-sso-oidc@3.572.0)': dependencies: - '@aws-sdk/client-sso-oidc': 3.572.0(@aws-sdk/client-sts@3.572.0) + '@aws-sdk/client-sso-oidc': 3.572.0 '@aws-sdk/types': 3.567.0 '@smithy/property-provider': 2.2.0 '@smithy/shared-ini-file-loader': 2.4.0