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..d496467b1a 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -10969,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 @@ -11164,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 @@ -11207,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': @@ -11383,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 @@ -11426,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': @@ -11591,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) @@ -11768,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 @@ -12080,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 diff --git a/services/apps/members_enrichment_worker/src/activities/enrichment.ts b/services/apps/members_enrichment_worker/src/activities/enrichment.ts index b646d5a759..36b7eb6c13 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 @@ -244,7 +247,7 @@ export async function getPriorityArray(): Promise { export async function fetchMemberDataForLLMSquashing( memberId: string, -): Promise { +): Promise { return fetchMemberDataForLLMSquashingDb(svc.postgres.reader.connection(), memberId) } @@ -601,12 +604,16 @@ 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. - // 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) { @@ -788,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[] { @@ -808,135 +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, -): 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, - ) - - 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..3b8945d3ae --- /dev/null +++ b/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.test.ts @@ -0,0 +1,322 @@ +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, + // on real data the resolved org lands on organizationId, not on identities — enrichment + // does not backfill identities with the org it just resolved + identities: [ + { + 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('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'])) + + 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('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('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('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' }) + 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..8820d18f5f --- /dev/null +++ b/services/apps/members_enrichment_worker/src/activities/workExperienceReconciliation.ts @@ -0,0 +1,280 @@ +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> +} + +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 + * 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 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) { + if (!createdKeys.has(key)) return true + } + 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 + * old rows are deleted, unmatched new entries are created. Never touches UI/project-registry + * or verified rows — callers must exclude those from oldEnrichmentRows and filter matching + * newEntries out beforehand. + */ +function reconcileEnrichmentOrgs( + oldEnrichmentRows: IMemberOrganizationData[], + newEntries: IMemberEnrichmentDataNormalizedOrganization[], +): IWorkExperienceChanges { + 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() + const pendingDateChanges: IPendingOrgUpdate[] = [] + + 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) && + normalizeDate(c.dateStart) === normalizeDate(entry.startDate) && + normalizeDate(c.dateEnd) === normalizeDate(entry.endDate), + ) ?? + 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 + } + const startChanged = normalizeDate(entry.startDate) !== normalizeDate(match.dateStart) + const endChanged = normalizeDate(entry.endDate) !== normalizeDate(match.dateEnd) + if (startChanged) { + toUpdateInner.dateStart = entry.startDate + } + if (endChanged) { + toUpdateInner.dateEnd = entry.endDate + } + if (entry.source !== undefined && entry.source !== match.source) { + toUpdateInner.source = entry.source + } + 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 } +} + +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)) + + // 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. 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.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.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.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/libs/data-access-layer/src/members/organizations.ts b/services/libs/data-access-layer/src/members/organizations.ts index 3a31a2a9d0..b907f47339 100644 --- a/services/libs/data-access-layer/src/members/organizations.ts +++ b/services/libs/data-access-layer/src/members/organizations.ts @@ -513,15 +513,19 @@ export async function deleteMemberOrganizations( memberId: string, ids?: string[], softDelete = true, + deletedBy?: string, ): Promise { - // Base query depends on soft vs hard delete + // 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 - ? '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 51ffd36fc5..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, @@ -57,7 +59,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."deletedBy" is not null) select m."displayName", m.attributes, m."manuallyChangedFields", @@ -84,13 +92,18 @@ 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 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 @@ -562,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 831de9d529..4d0581e674 100644 --- a/services/libs/types/src/enrichment.ts +++ b/services/libs/types/src/enrichment.ts @@ -40,9 +40,15 @@ export interface IMemberOrganizationData { dateStart: string dateEnd: string source: string + verified?: boolean + verifiedBy?: string | null identities?: IOrganizationIdentity[] } +export interface IDeletedMemberOrganizationData { + orgId: string +} + export interface IMemberOriginalData { // members table data displayName: string @@ -55,6 +61,9 @@ export interface IMemberOriginalData { // memberOrganizations table data organizations: IMemberOrganizationData[] + // 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 {