feat: add recalculate affiliations script

Signed-off-by: Umberto Sgueglia <usgueglia@contractor.linuxfoundation.org>

Author: ulemons

pnpm-lock.yaml

@@ -8,6 +8,7 @@
     "dev": "nodemon --watch src --watch ../../libs --ext ts --exec pnpm run start:debug",
     "cleanup-fork-activities": "npx tsx src/bin/cleanup-fork-activities-and-maintainers.ts",
     "cleanup-member-aggregates": "npx tsx src/bin/cleanup-member-aggregates.ts",
+    "recalculate-enrichment-affiliations": "npx tsx src/bin/recalculate-enrichment-affiliations.ts",
     "add-lf-projects-to-collection": "npx tsx src/bin/add-lf-projects-to-collection.ts",
     "lint": "npx eslint --ext .ts src --max-warnings=0",
     "format": "npx prettier --write \"src/**/*.ts\"",
@@ -24,6 +25,7 @@
     "@crowd/redis": "workspace:*",
     "@crowd/snowflake": "workspace:*",
     "@crowd/types": "workspace:*",
+    "@crowd/temporal": "workspace:*",
     "@temporalio/client": "~1.11.8",
     "@temporalio/workflow": "~1.11.8",
     "axios": "^1.6.8",

services/apps/script_executor_worker/package.json

@@ -0,0 +1,264 @@
+/**
+ * Recalculate Enrichment Affiliations Script
+ *
+ * PROBLEM:
+ * Before fix CM-1118, the members_enrichment_worker modified work experiences
+ * (memberOrganizations) without triggering affiliation recalculation. As a result,
+ * activityRelations.organizationId may be stale for members whose work experiences
+ * were created, updated, or deleted by enrichment.
+ *
+ * SOLUTION:
+ * This script finds all members with enrichment-sourced work experiences and triggers
+ * a memberUpdate Temporal workflow for each, which recalculates affiliations and syncs
+ * to OpenSearch.
+ *
+ * Usage:
+ *   # Via package.json script (recommended):
+ *   pnpm run recalculate-enrichment-affiliations -- [options]
+ *
+ *   # Or directly with tsx:
+ *   npx tsx src/bin/recalculate-enrichment-affiliations.ts [options]
+ *
+ * Options:
+ *   --page-size <n>      Number of members to fetch per DB page (default: 1000)
+ *   --concurrency <n>    Max concurrent Temporal workflow starts per page (default: 20)
+ *   --page-delay <ms>    Milliseconds to wait between pages (default: 5000)
+ *   --start-after <id>   Resume from a specific memberId (exclusive, for restarts)
+ *   --dry-run            Log what would be processed without starting workflows
+ *   --limit <n>          Stop after processing at most N members total (for testing)
+ *
+ * Environment Variables Required:
+ *   CROWD_DB_WRITE_HOST        - Postgres write host
+ *   CROWD_DB_PORT              - Postgres port
+ *   CROWD_DB_USERNAME          - Postgres username
+ *   CROWD_DB_PASSWORD          - Postgres password
+ *   CROWD_DB_DATABASE          - Postgres database name
+ *   CROWD_TEMPORAL_SERVER_URL  - Temporal server URL
+ *   CROWD_TEMPORAL_NAMESPACE   - Temporal namespace
+ *   SERVICE                    - Service identifier (used by Temporal client)
+ */
+import { DEFAULT_TENANT_ID } from '@crowd/common'
+import { WRITE_DB_CONFIG, getDbConnection } from '@crowd/data-access-layer/src/database'
+import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor'
+import { getServiceChildLogger } from '@crowd/logging'
+import { TEMPORAL_CONFIG, getTemporalClient } from '@crowd/temporal'
+import { WorkflowIdReusePolicy } from '@temporalio/client'
+
+const ENRICHMENT_SOURCES = ['enrichment-progai', 'enrichment-clearbit', 'enrichment-crustdata']
+
+const log = getServiceChildLogger('recalculate-enrichment-affiliations')
+
+interface MemberWithOrgs {
+  memberId: string
+  organizationIds: string[]
+}
+
+interface ScriptOptions {
+  pageSize: number
+  concurrency: number
+  pageDelayMs: number
+  startAfter: string | null
+  dryRun: boolean
+  limit: number | null
+}
+
+function parseArgs(): ScriptOptions {
+  const args = process.argv.slice(2)
+
+  const getArg = (flag: string): string | undefined => {
+    const idx = args.indexOf(flag)
+    if (idx !== -1 && idx + 1 < args.length) return args[idx + 1]
+    return undefined
+  }
+
+  const pageSize = parseInt(getArg('--page-size') ?? '1000', 10)
+  const concurrency = parseInt(getArg('--concurrency') ?? '20', 10)
+  const pageDelayMs = parseInt(getArg('--page-delay') ?? '5000', 10)
+  const startAfter = getArg('--start-after') ?? null
+  const dryRun = args.includes('--dry-run')
+  const limitRaw = getArg('--limit')
+  const limit = limitRaw !== undefined ? parseInt(limitRaw, 10) : null
+
+  if (isNaN(pageSize) || pageSize <= 0) {
+    log.error('--page-size must be a positive integer')
+    process.exit(1)
+  }
+  if (isNaN(concurrency) || concurrency <= 0) {
+    log.error('--concurrency must be a positive integer')
+    process.exit(1)
+  }
+  if (isNaN(pageDelayMs) || pageDelayMs < 0) {
+    log.error('--page-delay must be a non-negative integer')
+    process.exit(1)
+  }
+  if (limit !== null && (isNaN(limit) || limit <= 0)) {
+    log.error('--limit must be a positive integer')
+    process.exit(1)
+  }
+
+  return { pageSize, concurrency, pageDelayMs, startAfter, dryRun, limit }
+}
+
+async function fetchPage(
+  qx: ReturnType<typeof pgpQx>,
+  afterMemberId: string | null,
+  pageSize: number,
+): Promise<MemberWithOrgs[]> {
+  const cursorClause = afterMemberId ? `AND "memberId" > $(afterMemberId)` : ''
+
+  // Selects only members that have at least one work experience created by enrichment
+  // (source IN enrichment-progai, enrichment-clearbit, enrichment-crustdata).
+  // organizationIds contains only the enrichment-sourced org IDs for that member —
+  // used by the memberUpdate workflow to determine which orgs to sync to OpenSearch.
+  const rows = await qx.select(
+    `
+    SELECT "memberId", array_agg(DISTINCT "organizationId") AS "organizationIds"
+    FROM "memberOrganizations"
+    WHERE source = ANY($(sources))
+    ${cursorClause}
+    GROUP BY "memberId"
+    ORDER BY "memberId"
+    LIMIT $(pageSize)
+    `,
+    { sources: ENRICHMENT_SOURCES, afterMemberId, pageSize },
+  )
+
+  return rows as MemberWithOrgs[]
+}
+
+async function runWithConcurrency<T>(
+  items: T[],
+  concurrency: number,
+  fn: (item: T) => Promise<void>,
+): Promise<{ succeeded: number; failed: number }> {
+  let succeeded = 0
+  let failed = 0
+  let index = 0
+
+  async function worker() {
+    while (index < items.length) {
+      const item = items[index++]
+      try {
+        await fn(item)
+        succeeded++
+      } catch (err) {
+        failed++
+        log.error({ err }, 'Failed to process member')
+      }
+    }
+  }
+
+  const workers = Array.from({ length: Math.min(concurrency, items.length) }, worker)
+  await Promise.all(workers)
+
+  return { succeeded, failed }
+}
+
+async function main() {
+  const opts = parseArgs()
+
+  log.info('='.repeat(80))
+  log.info('Recalculate Enrichment Affiliations Script')
+  log.info('='.repeat(80))
+  log.info(`Page size:   ${opts.pageSize}`)
+  log.info(`Concurrency: ${opts.concurrency}`)
+  log.info(`Page delay:  ${opts.pageDelayMs}ms`)
+  log.info(`Start after: ${opts.startAfter ?? '(beginning)'}`)
+  log.info(`Mode:        ${opts.dryRun ? 'DRY RUN' : 'LIVE'}`)
+  log.info(`Limit:       ${opts.limit ?? '(none)'}`)
+  log.info('='.repeat(80))
+
+  // Init DB
+  const dbConnection = await getDbConnection(WRITE_DB_CONFIG())
+  const qx = pgpQx(dbConnection)
+
+  // Init Temporal
+  const temporal = await getTemporalClient(TEMPORAL_CONFIG())
+
+  let cursor: string | null = opts.startAfter
+  let pageNum = 0
+  let totalSucceeded = 0
+  let totalFailed = 0
+  let totalProcessed = 0
+
+  while (true) {
+    pageNum++
+
+    const remaining = opts.limit !== null ? opts.limit - totalProcessed : opts.pageSize
+    if (remaining <= 0) {
+      log.info(`Limit of ${opts.limit} members reached.`)
+      break
+    }
+
+    const fetchSize = Math.min(opts.pageSize, remaining)
+    const membersPage = await fetchPage(qx, cursor, fetchSize)
+
+    if (membersPage.length === 0) {
+      log.info('No more members to process.')
+      break
+    }
+
+    const lastMemberId = membersPage[membersPage.length - 1].memberId
+    log.info(`Page ${pageNum}: ${membersPage.length} members | cursor: ${lastMemberId}`)
+
+    if (opts.dryRun) {
+      log.info(`[DRY RUN] Would trigger ${membersPage.length} workflows`)
+      totalProcessed += membersPage.length
+    } else {
+      const { succeeded, failed } = await runWithConcurrency(
+        membersPage,
+        opts.concurrency,
+        async ({ memberId, organizationIds }) => {
+          await temporal.workflow.start('memberUpdate', {
+            taskQueue: 'profiles',
+            workflowId: `member-update/${DEFAULT_TENANT_ID}/${memberId}`,
+            workflowIdReusePolicy:
+              WorkflowIdReusePolicy.WORKFLOW_ID_REUSE_POLICY_TERMINATE_IF_RUNNING,
+            retry: { maximumAttempts: 10 },
+            args: [
+              {
+                member: { id: memberId },
+                memberOrganizationIds: organizationIds,
+                syncToOpensearch: true,
+              },
+            ],
+            searchAttributes: { TenantId: [DEFAULT_TENANT_ID] },
+          })
+        },
+      )
+
+      totalSucceeded += succeeded
+      totalFailed += failed
+      totalProcessed += succeeded + failed
+      log.info(`Page ${pageNum} done: ${succeeded} ok, ${failed} failed`)
+    }
+
+    log.info(`Resume with: --start-after ${lastMemberId}`)
+    cursor = lastMemberId
+
+    if (membersPage.length < fetchSize) {
+      // Last page (fewer results than requested means no more data)
+      break
+    }
+
+    if (opts.pageDelayMs > 0) {
+      await new Promise((resolve) => setTimeout(resolve, opts.pageDelayMs))
+    }
+  }
+
+  log.info('='.repeat(80))
+  log.info('Summary')
+  log.info('='.repeat(80))
+  log.info(`Pages processed: ${pageNum}`)
+  if (!opts.dryRun) {
+    log.info(`Total succeeded: ${totalSucceeded}`)
+    log.info(`Total failed:    ${totalFailed}`)
+  }
+
+  process.exit(totalFailed > 0 ? 1 : 0)
+}
+
+main().catch((err) => {
+  log.error({ err }, 'Unexpected error')
+  process.exit(1)
+})