Skip to content

Commit ccae610

Browse files
skwowetskwowet
authored
fix: prevent enrichment timeouts and activityRelations write churn (CM-1179) (#4098)
Signed-off-by: Yeganathan S <63534555+skwowet@users.noreply.github.com>
1 parent c98d68e commit ccae610

30 files changed

Lines changed: 301 additions & 296 deletions

File tree

‎backend/src/api/public/v1/members/project-affiliations/patchProjectAffiliation.ts‎

Lines changed: 9 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ import { z } from 'zod'
33

44
import { captureApiChange, memberEditAffiliationsAction } from '@crowd/audit-logs'
55
import { NotFoundError } from '@crowd/common'
6-
import { CommonMemberService } from '@crowd/common_services'
6+
import { signalMemberUpdate } from '@crowd/common_services'
77
import {
88
MemberField,
99
deleteAllMemberSegmentAffiliationsForProject,
@@ -75,6 +75,10 @@ export async function patchProjectAffiliation(req: Request, res: Response): Prom
7575
memberEditAffiliationsAction(memberId, async (captureOldState, captureNewState) => {
7676
captureOldState(existingAffiliations)
7777

78+
const oldOrgIds = existingAffiliations.map((a) => a.organizationId)
79+
const newOrgIds = affiliations.map((a) => a.organizationId)
80+
const orgIdsToRecalculate = [...new Set([...oldOrgIds, ...newOrgIds])]
81+
7882
await qx.tx(async (tx) => {
7983
await deleteAllMemberSegmentAffiliationsForProject(tx, memberId, projectId)
8084

@@ -91,13 +95,11 @@ export async function patchProjectAffiliation(req: Request, res: Response): Prom
9195
})),
9296
)
9397
}
98+
})
9499

95-
const oldOrgIds = existingAffiliations.map((a) => a.organizationId)
96-
const newOrgIds = affiliations.map((a) => a.organizationId)
97-
const orgIdsToRecalculate = [...new Set([...oldOrgIds, ...newOrgIds])]
98-
99-
const service = new CommonMemberService(tx, req.temporal, req.log)
100-
await service.startAffiliationRecalculation(memberId, orgIdsToRecalculate)
100+
// Signal after commit so the workflow sees persisted changes
101+
await signalMemberUpdate(req.temporal, memberId, {
102+
memberOrganizationIds: orgIdsToRecalculate,
101103
})
102104

103105
updatedAffiliations = await fetchMemberSegmentAffiliationsForProject(qx, memberId, projectId)

‎backend/src/api/public/v1/members/work-experiences/createMemberWorkExperience.ts‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ import {
88
NotFoundError,
99
sanitizeMemberOrganizationDateRange,
1010
} from '@crowd/common'
11-
import { CommonMemberService } from '@crowd/common_services'
11+
import { signalMemberUpdate } from '@crowd/common_services'
1212
import {
1313
MemberField,
1414
changeMemberOrganizationAffiliationOverrides,
@@ -105,9 +105,11 @@ export async function createMemberWorkExperience(req: Request, res: Response): P
105105
},
106106
])
107107
}
108+
})
108109

109-
const service = new CommonMemberService(tx, req.temporal, req.log)
110-
await service.startAffiliationRecalculation(memberId, [data.organizationId])
110+
// Signal after commit so the workflow sees persisted changes
111+
await signalMemberUpdate(req.temporal, memberId, {
112+
memberOrganizationIds: [data.organizationId],
111113
})
112114

113115
const orgsMap = await fetchManyMemberOrgsWithOrgData(qx, [memberId])

‎backend/src/api/public/v1/members/work-experiences/deleteMemberWorkExperience.ts‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ import { z } from 'zod'
33

44
import { captureApiChange, memberEditOrganizationsAction } from '@crowd/audit-logs'
55
import { NotFoundError } from '@crowd/common'
6-
import { CommonMemberService } from '@crowd/common_services'
6+
import { signalMemberUpdate } from '@crowd/common_services'
77
import {
88
MemberField,
99
deleteMemberOrganizations,
@@ -46,10 +46,11 @@ export async function deleteMemberWorkExperience(req: Request, res: Response): P
4646

4747
await qx.tx(async (tx) => {
4848
await deleteMemberOrganizations(tx, memberId, [workExperienceId])
49-
const commonMemberService = new CommonMemberService(tx, req.temporal, req.log)
50-
await commonMemberService.startAffiliationRecalculation(memberId, [
51-
memberOrg.organizationId,
52-
])
49+
})
50+
51+
// Signal after commit so the workflow sees persisted changes
52+
await signalMemberUpdate(req.temporal, memberId, {
53+
memberOrganizationIds: [memberOrg.organizationId],
5354
})
5455

5556
captureNewState(null)

‎backend/src/api/public/v1/members/work-experiences/updateMemberWorkExperience.ts‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ import { z } from 'zod'
33

44
import { captureApiChange, memberEditOrganizationsAction } from '@crowd/audit-logs'
55
import { BadRequestError, NotFoundError, sanitizeMemberOrganizationDateRange } from '@crowd/common'
6-
import { CommonMemberService } from '@crowd/common_services'
6+
import { signalMemberUpdate } from '@crowd/common_services'
77
import {
88
MemberField,
99
cleanSoftDeletedMemberOrganization,
@@ -80,9 +80,11 @@ export async function updateMemberWorkExperience(req: Request, res: Response): P
8080
await qx.tx(async (tx) => {
8181
await cleanSoftDeletedMemberOrganization(tx, memberId, data.organizationId, update)
8282
await updateMemberOrganization(tx, memberId, workExperienceId, update)
83+
})
8384

84-
const service = new CommonMemberService(tx, req.temporal, req.log)
85-
await service.startAffiliationRecalculation(memberId, [data.organizationId])
85+
// Signal after commit so the workflow sees persisted changes
86+
await signalMemberUpdate(req.temporal, memberId, {
87+
memberOrganizationIds: [data.organizationId],
8688
})
8789

8890
const orgsMap = await fetchManyMemberOrgsWithOrgData(qx, [memberId])

‎backend/src/api/public/v1/members/work-experiences/verifyMemberWorkExperience.ts‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ import { z } from 'zod'
33

44
import { captureApiChange, memberVerifyWorkExperienceAction } from '@crowd/audit-logs'
55
import { NotFoundError } from '@crowd/common'
6-
import { CommonMemberService } from '@crowd/common_services'
6+
import { signalMemberUpdate } from '@crowd/common_services'
77
import {
88
MemberField,
99
deleteMemberOrganizations,
@@ -62,12 +62,16 @@ export async function verifyMemberWorkExperience(req: Request, res: Response): P
6262
})
6363
} else {
6464
await deleteMemberOrganizations(tx, memberId, [workExperienceId], true)
65-
66-
const service = new CommonMemberService(tx, req.temporal, req.log)
67-
await service.startAffiliationRecalculation(memberId, [memberOrg.organizationId])
6865
}
6966
})
7067

68+
// Signal after commit so the workflow sees persisted changes
69+
if (!verified) {
70+
await signalMemberUpdate(req.temporal, memberId, {
71+
memberOrganizationIds: [memberOrg.organizationId],
72+
})
73+
}
74+
7175
captureNewState(updatedMemberOrg ?? { ...memberOrg, verified, verifiedBy })
7276
}),
7377
)

‎backend/src/bin/scripts/fix-members-activities-after-unaffilation.ts‎

Lines changed: 3 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
import commandLineArgs from 'command-line-args'
2-
import { randomUUID } from 'crypto'
32

3+
import { signalMemberUpdate } from '@crowd/common_services'
44
import { getDbConnection } from '@crowd/data-access-layer/src/database'
55
import { getServiceLogger } from '@crowd/logging'
66
import { getTemporalClient } from '@crowd/temporal'
@@ -99,26 +99,8 @@ setImmediate(async () => {
9999
try {
100100
log.info(`Processing member: ${member.id}`)
101101

102-
const uuid = randomUUID()
103-
104-
await temporal.workflow.start('memberUpdate', {
105-
taskQueue: 'profiles',
106-
workflowId: `member-update-fix-unaffiliation/${organizationId}/${member.id}/${uuid}`,
107-
retry: {
108-
maximumAttempts: 10,
109-
},
110-
args: [
111-
{
112-
member: {
113-
id: member.id,
114-
},
115-
memberOrganizationIds: [organizationId],
116-
syncToOpensearch: false,
117-
},
118-
],
119-
searchAttributes: {
120-
TenantId: ['875c38bd-2b1b-4e91-ad07-0cfbabb4c49f'], // default tenantId
121-
},
102+
await signalMemberUpdate(temporal, member.id, {
103+
memberOrganizationIds: [organizationId],
122104
})
123105

124106
processedCount++

‎backend/src/services/member/memberAffiliationsService.ts‎

Lines changed: 3 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,7 @@
22
import { uniq } from 'lodash'
33

44
import { Error400, dateIntersects, groupBy } from '@crowd/common'
5-
import { CommonMemberService } from '@crowd/common_services'
6-
import { optionsQx } from '@crowd/data-access-layer'
5+
import { signalMemberUpdate } from '@crowd/common_services'
76
import { findMaintainerRoles } from '@crowd/data-access-layer/src/maintainers'
87
import { fetchManySegments } from '@crowd/data-access-layer/src/segments'
98
import { LoggerBase } from '@crowd/logging'
@@ -108,12 +107,8 @@ export default class MemberAffiliationsService extends LoggerBase {
108107
this.options,
109108
)
110109

111-
const commonMemberService = new CommonMemberService(
112-
optionsQx(this.options),
113-
this.options.temporal,
114-
this.options.log,
115-
)
116-
await commonMemberService.startAffiliationRecalculation(data.memberId, [])
110+
await signalMemberUpdate(this.options.temporal, data.memberId)
111+
117112
return override
118113
}
119114
}

‎backend/src/services/member/memberOrganizationsService.ts‎

Lines changed: 20 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ import lodash from 'lodash'
33
import { Transaction } from 'sequelize'
44

55
import { Error404, sanitizeMemberOrganizationDateRange } from '@crowd/common'
6-
import { CommonMemberService } from '@crowd/common_services'
6+
import { signalMemberUpdate } from '@crowd/common_services'
77
import {
88
OrganizationField,
99
changeMemberOrganizationAffiliationOverrides,
@@ -14,7 +14,6 @@ import {
1414
fetchMemberOrganizationById,
1515
fetchMemberOrganizations,
1616
findMemberAffiliationOverrides,
17-
optionsQx,
1817
queryOrgs,
1918
updateMemberOrganization,
2019
} from '@crowd/data-access-layer'
@@ -36,16 +35,9 @@ type IOrganizationSummary = Pick<IOrganization, 'id' | 'displayName' | 'logo' |
3635
export default class MemberOrganizationsService extends LoggerBase {
3736
options: IServiceOptions
3837

39-
private readonly commonMemberService: CommonMemberService
40-
4138
constructor(options: IServiceOptions) {
4239
super(options.log)
4340
this.options = options
44-
this.commonMemberService = new CommonMemberService(
45-
optionsQx(options),
46-
options.temporal,
47-
options.log,
48-
)
4941
}
5042

5143
// Member organization list
@@ -196,13 +188,16 @@ export default class MemberOrganizationsService extends LoggerBase {
196188
])
197189
}
198190

199-
// Start affiliation recalculation within the same transaction
200-
await this.commonMemberService.startAffiliationRecalculation(memberId, [data.organizationId])
201-
202191
// Fetch updated list
203192
const result = await this.list(memberId, transaction)
204193

205194
await SequelizeRepository.commitTransaction(transaction)
195+
196+
// Signal after commit so the workflow sees persisted changes
197+
await signalMemberUpdate(this.options.temporal, memberId, {
198+
memberOrganizationIds: [data.organizationId],
199+
})
200+
206201
return result
207202
} catch (error) {
208203
await SequelizeRepository.rollbackTransaction(transaction)
@@ -259,11 +254,15 @@ export default class MemberOrganizationsService extends LoggerBase {
259254
new Set([existing.organizationId, data.organizationId]),
260255
).filter((orgId): orgId is string => Boolean(orgId))
261256

262-
await this.commonMemberService.startAffiliationRecalculation(memberId, orgsToRecalculate)
263-
264257
const result = await this.list(memberId, transaction)
265258

266259
await SequelizeRepository.commitTransaction(transaction)
260+
261+
// Signal after commit so the workflow sees persisted changes
262+
await signalMemberUpdate(this.options.temporal, memberId, {
263+
memberOrganizationIds: orgsToRecalculate,
264+
})
265+
267266
return result
268267
} catch (error) {
269268
await SequelizeRepository.rollbackTransaction(transaction)
@@ -288,15 +287,16 @@ export default class MemberOrganizationsService extends LoggerBase {
288287

289288
await deleteMemberOrganizations(qx, memberId, [id], true)
290289

291-
await this.commonMemberService.startAffiliationRecalculation(
292-
memberId,
293-
[memberOrganizationToBeDeleted.organizationId],
294-
true,
295-
)
296-
297290
const result = await this.list(memberId, transaction)
298291

299292
await SequelizeRepository.commitTransaction(transaction)
293+
294+
// Signal after commit so the workflow sees persisted changes
295+
await signalMemberUpdate(this.options.temporal, memberId, {
296+
memberOrganizationIds: [memberOrganizationToBeDeleted.organizationId],
297+
syncToOpensearch: true,
298+
})
299+
300300
return result
301301
} catch (error) {
302302
await SequelizeRepository.rollbackTransaction(transaction)

‎backend/src/services/memberService.ts‎

Lines changed: 4 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ import {
1010
getGithubInstallationToken,
1111
invalidateMemberQueryCache,
1212
prepareMemberUnmerge,
13+
signalMemberUpdate,
1314
startMemberUnmergeWorkflow,
1415
unmergeMember,
1516
} from '@crowd/common_services'
@@ -819,16 +820,10 @@ export default class MemberService extends LoggerBase {
819820
// Pass invalidateCache from options to control whether to clear list caches
820821
await invalidateMemberQueryCache(this.options.redis, [id], invalidateCache)
821822

822-
const commonMemberService = new CommonMemberService(
823-
optionsQx(this.options),
824-
this.options.temporal,
825-
this.options.log,
826-
)
827-
await commonMemberService.startAffiliationRecalculation(
828-
id,
829-
(data.organizations || []).map((o) => o.id),
823+
await signalMemberUpdate(this.options.temporal, id, {
824+
memberOrganizationIds: (data.organizations || []).map((o) => o.id),
830825
syncToOpensearch,
831-
)
826+
})
832827

833828
return record
834829
} catch (error) {

0 commit comments

Comments
 (0)