diff --git a/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts b/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts index 369fc4d6bb..0ebd5ef21e 100644 --- a/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts +++ b/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts @@ -15,17 +15,28 @@ import { fetchManyOrganizationAffiliationPolicies, fetchMemberOrganizationsBySource, findMemberById, + findOrgsByIds, updateMemberOrganization, } from '@crowd/data-access-layer' import { WRITE_DB_CONFIG, getDbConnection } from '@crowd/data-access-layer/src/database' import { deleteMemberSegmentAffiliations } from '@crowd/data-access-layer/src/member_segment_affiliations' +import { findMergedPrimaryIds } from '@crowd/data-access-layer/src/mergeActions/repo' import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor' +import { Logger } from '@crowd/logging' import { REDIS_CONFIG, RedisCache, RedisClient, getRedisClient } from '@crowd/redis' import { TEMPORAL_CONFIG, getTemporalClient } from '@crowd/temporal' -import { MemberOrgDate, MemberOrgStintChange, OrganizationSource } from '@crowd/types' +import { + IMemberOrganization, + MemberOrgDate, + MemberOrgStintChange, + MergeActionType, + OrganizationSource, +} from '@crowd/types' import { IJobDefinition } from '../types' +const MAX_FK_VIOLATION_RETRIES = 3 + const job: IJobDefinition = { name: 'infer-member-organization-stint-changes', cronTime: CronTime.every(5).minutes(), @@ -78,7 +89,19 @@ const job: IJobDefinition = { { withDeleted: true }, ) - const changes = inferMemberOrganizationStintChanges(memberId, existingOrgs, orgDates) + const reconciledOrgDates = await reconcileOrganizationDates( + qx, + existingOrgs, + orgDates, + memberId, + ctx.log, + ) + + const changes = inferMemberOrganizationStintChanges( + memberId, + existingOrgs, + reconciledOrgDates, + ) if (changes.length > 0) { ctx.log.debug({ memberId, changes }, 'Stint changes identified.') @@ -101,9 +124,32 @@ const job: IJobDefinition = { memberId, rawMembers, ) + await redis.del(fkViolationRetryKey(memberId)) processed++ } catch (err) { + if ((err as { code?: string })?.code === '23503') { + const constraint = (err as { constraint?: string })?.constraint + const retryKey = fkViolationRetryKey(memberId) + const retries = await redis.incr(retryKey) + + if (retries >= MAX_FK_VIOLATION_RETRIES) { + ctx.log.error( + err, + { memberId, constraint, retries }, + 'Stint change repeatedly referenced a missing related record; purging poisoned queue entry.', + ) + await purgeMember(redis, memberId) + } else { + ctx.log.warn( + err, + { memberId, constraint, retries }, + 'Stint change referenced a missing related record, will retry.', + ) + } + continue + } + ctx.log.error(err, { memberId }, 'Failed to process member stint inference.') throw err } @@ -129,19 +175,78 @@ function parseSetMembers(members: string[]): MemberOrgDate[] { return results } +// Merged orgs are rewritten to their primary id instead of dropped; only genuinely +// deleted orgs are dropped, since those can never resolve to a valid target. +async function reconcileOrganizationDates( + qx: QueryExecutor, + existingOrgs: IMemberOrganization[], + orgDates: MemberOrgDate[], + memberId: string, + log: Logger, +): Promise { + const knownOrgIds = new Set(existingOrgs.map((o) => o.organizationId)) + const orgIdsToVerify = [...new Set(orgDates.map((d) => d.organizationId))].filter( + (id) => !knownOrgIds.has(id), + ) + + if (orgIdsToVerify.length === 0) { + return orgDates + } + + const existingOrgIds = new Set((await findOrgsByIds(qx, orgIdsToVerify)).map((o) => o.id)) + const missingOrgIds = orgIdsToVerify.filter((id) => !existingOrgIds.has(id)) + + if (missingOrgIds.length === 0) { + return orgDates + } + + const mergedPrimaryIds = await findMergedPrimaryIds(qx, MergeActionType.ORG, missingOrgIds) + const deletedOrgIds = new Set(missingOrgIds.filter((id) => !mergedPrimaryIds.has(id))) + + if (deletedOrgIds.size > 0) { + log.warn( + { memberId, deletedOrgIds: [...deletedOrgIds] }, + 'Dropping queued stint dates for organizations that no longer exist.', + ) + } + + return orgDates + .filter((d) => !deletedOrgIds.has(d.organizationId)) + .map((d) => + mergedPrimaryIds.has(d.organizationId) + ? { ...d, organizationId: mergedPrimaryIds.get(d.organizationId) } + : d, + ) +} + /** * Purges a member from the queue and their associated Redis entries. */ async function purgeMember(redis: RedisClient, memberId: string): Promise { const datesKey = `${MEMBER_ORG_STINT_CHANGES_DATES_PREFIX}:${memberId}` + const retryKey = fkViolationRetryKey(memberId) + + await redis + .multi() + .del(datesKey) + .del(retryKey) + .sRem(MEMBER_ORG_STINT_CHANGES_QUEUE, memberId) + .exec() +} - await redis.multi().del(datesKey).sRem(MEMBER_ORG_STINT_CHANGES_QUEUE, memberId).exec() +function fkViolationRetryKey(memberId: string): string { + return `${MEMBER_ORG_STINT_CHANGES_DATES_PREFIX}:fk-violation-retries:${memberId}` } /** * Applies the stint changes to the database. */ async function applyStintChanges(qx: QueryExecutor, changes: MemberOrgStintChange[]) { + const insertOrgIds = [ + ...new Set(changes.filter((c) => c.type === 'insert').map((c) => c.organizationId)), + ] + const orgAffiliationPolicies = await fetchManyOrganizationAffiliationPolicies(qx, insertOrgIds) + for (const change of changes) { if (change.type === 'insert') { const memberOrganizationId = await createMemberOrganization(qx, change.memberId, { @@ -151,10 +256,6 @@ async function applyStintChanges(qx: QueryExecutor, changes: MemberOrgStintChang source: OrganizationSource.EMAIL_DOMAIN, }) - const orgAffiliationPolicies = await fetchManyOrganizationAffiliationPolicies(qx, [ - change.organizationId, - ]) - if (memberOrganizationId && orgAffiliationPolicies.get(change.organizationId)) { await changeMemberOrganizationAffiliationOverrides(qx, [ { diff --git a/services/libs/common_services/src/services/member-organization.ts b/services/libs/common_services/src/services/member-organization.ts index 5c2d9083f1..7e53b4302b 100644 --- a/services/libs/common_services/src/services/member-organization.ts +++ b/services/libs/common_services/src/services/member-organization.ts @@ -175,6 +175,10 @@ export function inferMemberOrganizationStintChanges( const activeRows = normalizedRows.filter((row) => !row.deletedAt) + const tombstonedOrgIds = new Set( + normalizedRows.filter((row) => row.deletedAt && row.deletedBy).map((row) => row.organizationId), + ) + // Deleted dated rows suppress recreation for dates the user removed const deletedRows = normalizedRows.filter( (row): row is typeof row & { dateStart: string } => !!row.deletedAt && !!row.dateStart, @@ -199,6 +203,10 @@ export function inferMemberOrganizationStintChanges( })) for (const { organizationId, date: targetDate } of sortedDates) { + if (tombstonedOrgIds.has(organizationId)) { + continue + } + if ( deletedRows.some( (row) => diff --git a/services/libs/data-access-layer/src/members/organizations.ts b/services/libs/data-access-layer/src/members/organizations.ts index e8b1f46432..de6346fe57 100644 --- a/services/libs/data-access-layer/src/members/organizations.ts +++ b/services/libs/data-access-layer/src/members/organizations.ts @@ -83,7 +83,8 @@ export async function fetchMemberOrganizationsBySource( "title", "memberId", "source", - "deletedAt" + "deletedAt", + "deletedBy" FROM "memberOrganizations" WHERE "memberId" = $(memberId) AND "source" = $(source) diff --git a/services/libs/data-access-layer/src/mergeActions/repo.ts b/services/libs/data-access-layer/src/mergeActions/repo.ts index 629ca69c26..bfc0324b60 100644 --- a/services/libs/data-access-layer/src/mergeActions/repo.ts +++ b/services/libs/data-access-layer/src/mergeActions/repo.ts @@ -80,6 +80,29 @@ export async function findEntityMergeActions( return result } +export async function findMergedPrimaryIds( + qx: QueryExecutor, + type: MergeActionType, + secondaryIds: string[], +): Promise> { + if (!secondaryIds.length) { + return new Map() + } + + const rows = await qx.select( + ` + SELECT ma."secondaryId", ma."primaryId" + FROM "mergeActions" ma + WHERE ma.type = $(type) + AND ma.state = $(state) + AND ma."secondaryId" = ANY($(secondaryIds)::uuid[]) + `, + { type, state: MergeActionState.MERGED, secondaryIds }, + ) + + return new Map(rows.map((r) => [r.secondaryId, r.primaryId])) +} + export async function setMergeAction( qx: QueryExecutor, type: MergeActionType, diff --git a/services/libs/types/src/organizations.ts b/services/libs/types/src/organizations.ts index 78e74f4ed5..abd2dd5ead 100644 --- a/services/libs/types/src/organizations.ts +++ b/services/libs/types/src/organizations.ts @@ -63,6 +63,7 @@ export interface IMemberOrganization { verified?: boolean verifiedBy?: string deletedAt?: string + deletedBy?: string displayName?: string affiliationOverride?: IMemberOrganizationAffiliationOverride }