feat: infer memberOrganization stint dates from work-email activities (CM-1105) (#4054)

Author: skwowet

backend/src/database/migrations/V1776931245__member-organizations-email-domain-partial-index.sql

@@ -0,0 +1,3 @@
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "ix_memberOrganizations_memberId_emailDomain"
+  ON "memberOrganizations" ("memberId")
+  WHERE "source" = 'email-domain' AND "deletedAt" IS NULL;

backend/src/services/member/memberOrganizationsService.ts

@@ -22,6 +22,7 @@ import {
   IOrganization,
   IRenderFriendlyMemberOrganization,
   MemberOrganizationUpdate,
+  OrganizationSource,
 } from '@crowd/types'
 
 import SequelizeRepository from '@/database/repositories/sequelizeRepository'
@@ -219,14 +220,18 @@ export default class MemberOrganizationsService extends LoggerBase {
           title: data.title,
           dateStart: data.dateStart,
           dateEnd: data.dateEnd,
-          source: data.source,
           verified: data.verified,
           verifiedBy: data.verifiedBy,
         }).filter(([, v]) => v !== undefined),
       )
 
       await cleanSoftDeletedMemberOrganization(qx, memberId, data.organizationId, data)
-      await updateMemberOrganization(qx, memberId, id, update)
+      // Any manual edit from the frontend promotes ownership to UI so automated
+      // sources (e.g. email-domain inference) no longer overwrite user intent.
+      await updateMemberOrganization(qx, memberId, id, {
+        ...update,
+        source: OrganizationSource.UI,
+      })
 
       await this.commonMemberService.startAffiliationRecalculation(memberId, [data.organizationId])
 

services/apps/cron_service/src/jobs/inferMemberOrganizationStintChanges.job.ts

@@ -0,0 +1,141 @@
+import CronTime from 'cron-time-generator'
+
+import {
+  MEMBER_ORG_STINT_CHANGES_DATES_PREFIX,
+  MEMBER_ORG_STINT_CHANGES_QUEUE,
+  inferMemberOrganizationStintChanges,
+} from '@crowd/common_services'
+import {
+  QueryExecutor,
+  changeMemberOrganizationAffiliationOverrides,
+  checkOrganizationAffiliationPolicy,
+  createMemberOrganization,
+  fetchMemberOrganizationsBySource,
+  updateMemberOrganization,
+} from '@crowd/data-access-layer'
+import { WRITE_DB_CONFIG, getDbConnection } from '@crowd/data-access-layer/src/database'
+import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor'
+import { REDIS_CONFIG, RedisCache, getRedisClient } from '@crowd/redis'
+import { MemberOrgDate, MemberOrgStintChange, OrganizationSource } from '@crowd/types'
+
+import { IJobDefinition } from '../types'
+
+const job: IJobDefinition = {
+  name: 'infer-member-organization-stint-changes',
+  cronTime: CronTime.every(5).minutes(),
+  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
+
+    ctx.log.info({ count: memberIds.length }, 'Processing members from queue.')
+
+    let processed = 0
+
+    for (const memberId of memberIds) {
+      try {
+        const datesKey = `${MEMBER_ORG_STINT_CHANGES_DATES_PREFIX}:${memberId}`
+        const rawMembers = await redis.sMembers(datesKey)
+
+        if (!rawMembers?.length) {
+          await redis.sRem(MEMBER_ORG_STINT_CHANGES_QUEUE, memberId)
+          continue
+        }
+
+        const orgDates = parseSetMembers(rawMembers)
+
+        if (orgDates.length > 0) {
+          const existingOrgs = await fetchMemberOrganizationsBySource(
+            qx,
+            memberId,
+            OrganizationSource.EMAIL_DOMAIN,
+          )
+
+          const changes = inferMemberOrganizationStintChanges(memberId, existingOrgs, orgDates)
+
+          if (changes.length > 0) {
+            ctx.log.debug({ memberId, changes }, 'Stint changes identified.')
+            await qx.tx((tx) => applyStintChanges(tx, changes))
+          }
+        }
+
+        // Atomically remove only the values we read.
+        // If no new values were added, remove the member from the queue.
+        await RedisCache.ackSetMembers(
+          redis,
+          datesKey,
+          MEMBER_ORG_STINT_CHANGES_QUEUE,
+          memberId,
+          rawMembers,
+        )
+
+        processed++
+      } catch (err) {
+        ctx.log.error(err, { memberId }, 'Failed to process member stint inference.')
+        throw err
+      }
+    }
+
+    ctx.log.info({ processed }, 'Batch complete.')
+  },
+}
+
+/**
+ * Parses set members of the form "orgId|date" into typed activity dates.
+ */
+function parseSetMembers(members: string[]): MemberOrgDate[] {
+  const results: MemberOrgDate[] = []
+
+  for (const m of members) {
+    const idx = m.indexOf('|')
+    if (idx > 0) {
+      results.push({ organizationId: m.slice(0, idx), date: m.slice(idx + 1) })
+    }
+  }
+
+  return results
+}
+
+/**
+ * Applies the stint changes to the database.
+ */
+async function applyStintChanges(qx: QueryExecutor, changes: MemberOrgStintChange[]) {
+  for (const change of changes) {
+    if (change.type === 'insert') {
+      const memberOrganizationId = await createMemberOrganization(qx, change.memberId, {
+        organizationId: change.organizationId,
+        dateStart: change.dateStart,
+        dateEnd: change.dateEnd,
+        source: OrganizationSource.EMAIL_DOMAIN,
+      })
+
+      const isAffiliationBlocked = await checkOrganizationAffiliationPolicy(
+        qx,
+        change.organizationId,
+      )
+
+      if (memberOrganizationId && isAffiliationBlocked) {
+        await changeMemberOrganizationAffiliationOverrides(qx, [
+          {
+            memberId: change.memberId,
+            memberOrganizationId,
+            allowAffiliation: false,
+          },
+        ])
+      }
+    } else {
+      await updateMemberOrganization(qx, change.memberId, change.id, {
+        dateStart: change.dateStart,
+        dateEnd: change.dateEnd,
+      })
+    }
+  }
+}
+
+export default job

services/apps/data_sink_worker/package.json

@@ -17,8 +17,6 @@
     "script:restart-result": "SERVICE=script tsx src/bin/restart-result.ts",
     "script:process-results": "SERVICE=script tsx src/bin/process-results.ts",
     "script:trigger-results-for-tenant": "SERVICE=script tsx src/bin/trigger-results-for-tenant.ts",
-    "script:map-tenant-members-to-org": "SERVICE=script tsx src/bin/map-tenant-members-to-org.ts",
-    "script:map-member-to-org": "SERVICE=script tsx src/bin/map-member-to-org.ts",
     "script:fix-activity-obj-member-data": "SERVICE=script tsx src/bin/fix-activity-obj-member-data.ts",
     "script:fix-member-displayName": "SERVICE=script tsx src/bin/fix-member-displayName.ts",
     "script:fix-members-joinedAt": "SERVICE=script tsx src/bin/fix-members-joinedAt.ts",