refactor: process member affiliations sequentially within a trnx

services/apps/script_executor_worker/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@crowd/script-executor-worker",
   "scripts": {
-    "start": "CROWD_TEMPORAL_TASKQUEUE=script-executor SERVICE=script-executor-worker tsx src/main.ts",
+    "start": "CROWD_TEMPORAL_TASKQUEUE=script-executor SERVICE=script-executor-worker LOG_LEVEL=trace tsx src/main.ts",
     "start:debug:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && CROWD_TEMPORAL_TASKQUEUE=script-executor SERVICE=script-executor-worker LOG_LEVEL=trace tsx --inspect=0.0.0.0:9232 src/main.ts",
     "start:debug": "CROWD_TEMPORAL_TASKQUEUE=script-executor SERVICE=script-executor-worker LOG_LEVEL=info tsx --inspect=0.0.0.0:9232 src/main.ts",
     "dev:local": "nodemon --watch src --watch ../../libs --ext ts --exec pnpm run start:debug:local",

services/apps/script_executor_worker/src/types.ts

@@ -61,3 +61,7 @@ export interface IBlockOrganizationAffiliationArgs {
   organizationId: string
   offset?: number
 }
+
+export interface IRefreshMemberAffiliationsArgs extends IScriptBatchTestArgs {
+  memberIds?: string[]
+}

services/apps/script_executor_worker/src/workflows.ts

@@ -8,6 +8,7 @@ import { findAndMergeMembersWithSameVerifiedEmailsInDifferentPlatforms } from '.
 import { fixBotMembersAffiliation } from './workflows/fix-bot-members-affiliation'
 import { fixOrgIdentitiesWithWrongUrls } from './workflows/fixOrgIdentitiesWithWrongUrls'
 import { processLLMVerifiedMerges } from './workflows/processLLMVerifiedMerges'
+import { refreshMemberAffiliations } from './workflows/refresh-member-affiliations'
 import { syncMembers } from './workflows/sync/members'
 import { syncOrganizations } from './workflows/sync/organizations'
 
@@ -24,4 +25,5 @@ export {
   cleanupDuplicateMembers,
   fixBotMembersAffiliation,
   blockOrganizationAffiliation,
+  refreshMemberAffiliations,
 }

services/apps/script_executor_worker/src/workflows/refresh-member-affiliations.ts

@@ -0,0 +1,24 @@
+import { proxyActivities } from '@temporalio/workflow'
+
+import * as activities from '../activities'
+import { IRefreshMemberAffiliationsArgs } from '../types'
+
+const { calculateMemberAffiliations } = proxyActivities<typeof activities>({
+  startToCloseTimeout: '30 minutes',
+  retry: { maximumAttempts: 3, backoffCoefficient: 3 },
+})
+
+export async function refreshMemberAffiliations(
+  args: IRefreshMemberAffiliationsArgs,
+): Promise<void> {
+  const memberIds = args.memberIds ?? []
+
+  if (memberIds.length === 0) {
+    console.log('No member ids to refresh!')
+    return
+  }
+
+  for (const memberId of memberIds) {
+    await calculateMemberAffiliations(memberId)
+  }
+}

services/libs/data-access-layer/src/member-organization-affiliation/index.ts

@@ -337,6 +337,7 @@ async function processAffiliationActivities(
         WHERE "activityId" in (
           select "activityId" from "activityRelations"
           where ${whereClause}
+          order by "timestamp" asc, "activityId" asc
           limit $(batchSize)
         )
       `,
@@ -357,13 +358,18 @@ export async function refreshMemberOrganizationAffiliations(qx: QueryExecutor, m
 
   const affiliations = await prepareMemberOrganizationAffiliationTimeline(qx, memberId)
 
-  // process timeline in parallel
-  const results = await Promise.all(
-    affiliations.map((affiliation) => processAffiliationActivities(qx, memberId, affiliation)),
-  )
+  logger.debug({ affiliations }, 'Member organization affiliations timeline!')
+
+  let processed = 0
+
+  await qx.tx(async (tx) => {
+    // process timeline sequentially to avoid race conditions
+    for (const affiliation of affiliations) {
+      processed += await processAffiliationActivities(tx, memberId, affiliation)
+    }
+  })
 
   const duration = performance.now() - start
-  const processed = results.reduce((acc, processed) => acc + processed, 0)
 
   logger.info({ memberId }, `Refreshed ${processed} activities in ${duration}ms`)
 }