diff --git a/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts b/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts index 6ee74b31a9..0ececd8b1b 100644 --- a/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts +++ b/services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts @@ -4,6 +4,7 @@ import { MEMBER_ORG_STINT_CHANGES_DATES_PREFIX, MEMBER_ORG_STINT_CHANGES_QUEUE, inferMemberOrganizationStintChanges, + signalMemberUpdate, } from '@crowd/common_services' import { QueryExecutor, @@ -18,6 +19,7 @@ import { WRITE_DB_CONFIG, getDbConnection } from '@crowd/data-access-layer/src/d import { deleteMemberSegmentAffiliations } from '@crowd/data-access-layer/src/member_segment_affiliations' import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor' import { REDIS_CONFIG, RedisCache, getRedisClient } from '@crowd/redis' +import { TEMPORAL_CONFIG, getTemporalClient } from '@crowd/temporal' import { MemberOrgDate, MemberOrgStintChange, OrganizationSource } from '@crowd/types' import { IJobDefinition } from '../types' @@ -28,14 +30,17 @@ const job: IJobDefinition = { timeout: 10 * 60, process: async (ctx) => { const redis = await getRedisClient(REDIS_CONFIG()) - const db = await getDbConnection(WRITE_DB_CONFIG()) - const qx = pgpQx(db) ctx.log.info('Starting member organization stint inference job.') const memberIds = await redis.sRandMemberCount(MEMBER_ORG_STINT_CHANGES_QUEUE, 500) + if (!memberIds?.length) return + const db = await getDbConnection(WRITE_DB_CONFIG()) + const qx = pgpQx(db) + const temporal = await getTemporalClient(TEMPORAL_CONFIG()) + ctx.log.info({ count: memberIds.length }, 'Processing members from queue.') let processed = 0 @@ -64,6 +69,12 @@ const job: IJobDefinition = { if (changes.length > 0) { ctx.log.debug({ memberId, changes }, 'Stint changes identified.') await qx.tx((tx) => applyStintChanges(tx, changes)) + + ctx.log.debug( + { memberId }, + 'Triggering member update workflow to refresh affiliations.', + ) + await signalMemberUpdate(temporal, memberId) } }