Merge branch 'improve/CM-1105' into script/CM-1107

Author: skwowet

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

@@ -15,8 +15,8 @@ import {
 } 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, getRedisClient } from '@crowd/redis'
-import { MemberOrgStintChange, OrganizationSource } from '@crowd/types'
+import { REDIS_CONFIG, RedisCache, getRedisClient } from '@crowd/redis'
+import { MemberOrgDate, MemberOrgStintChange, OrganizationSource } from '@crowd/types'
 
 import { IJobDefinition } from '../types'
 
@@ -41,40 +41,44 @@ const job: IJobDefinition = {
     for (const memberId of memberIds) {
       try {
         const datesKey = `${MEMBER_ORG_STINT_CHANGES_DATES_PREFIX}:${memberId}`
-        const hash = await redis.hGetAll(datesKey)
+        const rawMembers = await redis.sMembers(datesKey)
 
-        if (!hash || Object.keys(hash).length === 0) {
+        if (!rawMembers?.length) {
           await redis.sRem(MEMBER_ORG_STINT_CHANGES_QUEUE, memberId)
           continue
         }
 
-        const { activityDates, orgIds } = parseMemberActivityHash(hash)
+        const orgDates = parseSetMembers(rawMembers)
 
-        if (activityDates.length > 0) {
+        if (orgDates.length > 0) {
           const existingOrgs = await fetchMemberOrganizationsBySource(
             qx,
             memberId,
             OrganizationSource.EMAIL_DOMAIN,
           )
 
-          const changes = inferMemberOrganizationStintChanges(memberId, existingOrgs, activityDates)
+          const changes = inferMemberOrganizationStintChanges(memberId, existingOrgs, orgDates)
 
           if (changes.length > 0) {
             ctx.log.debug({ memberId, changes }, 'Stint changes identified.')
-            await applyStintChanges(qx, changes)
+            await qx.tx((tx) => applyStintChanges(tx, changes))
           }
         }
 
-        // Remove only the fields we actually read
-        await redis
-          .multi()
-          .hDel(datesKey, orgIds)
-          .sRem(MEMBER_ORG_STINT_CHANGES_QUEUE, memberId)
-          .exec()
+        // 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
       }
     }
 
@@ -83,23 +87,19 @@ const job: IJobDefinition = {
 }
 
 /**
- * Parses the Redis hash into a clean, typed list of activity dates.
+ * Parses set members of the form "orgId|date" into typed activity dates.
  */
-function parseMemberActivityHash(hash: Record<string, string>) {
-  const orgIds = Object.keys(hash)
-  const activityDates = orgIds.flatMap((organizationId) => {
-    try {
-      const dates = JSON.parse(hash[organizationId])
-      return Array.isArray(dates)
-        ? dates
-            .filter((d): d is string => typeof d === 'string')
-            .map((date) => ({ organizationId, date }))
-        : []
-    } catch {
-      return []
+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 { activityDates, orgIds }
+  }
+
+  return results
 }
 
 /**

services/apps/data_sink_worker/src/service/member.service.ts

@@ -1065,56 +1065,26 @@ export default class MemberService extends LoggerBase {
   }
 
   /**
-   * Queues member activity dates in Redis as raw input for stint inference.
-   *
-   * To maximize throughput, this uses a non-atomic HGET/HSET pattern. While
-   * concurrent writes may occasionally drop a date, the system is self-healing
-   * as future activity will re-populate the buffer.
+   * Queues a member activity date in Redis for stint inference.
+   * Uses SADD for natural concurrency safety and deduplication.
    */
   private async bufferMemberOrganizationActivityDates(
     memberId: string,
     organizationId: string,
     activityTimestamp: string,
   ): Promise<void> {
     const date = new Date(activityTimestamp).toISOString().split('T')[0]
-    const key = `${MEMBER_ORG_STINT_CHANGES_DATES_PREFIX}:${memberId}`
-
-    // Safe parser for existing Redis strings.
-    // Returns a valid string array even if data is corrupt or missing.
-    const parseExistingDates = (value: string | null | undefined): string[] => {
-      if (!value) return []
-      try {
-        const parsed = JSON.parse(value)
-        return Array.isArray(parsed) ? parsed.filter((d): d is string => typeof d === 'string') : []
-      } catch {
-        this.log.warn(
-          { memberId, organizationId, key },
-          'Corrupt dates buffer value detected during buffering for member organization activity dates.',
-        )
-        return []
-      }
-    }
 
-    // 1. Fetch current dates for this specific organization
-    const existing = await this.redisClient.hGet(key, organizationId)
-    const dates: string[] = parseExistingDates(existing)
+    // Each member gets one flat set: values are "orgId|date"
+    const datesKey = `${MEMBER_ORG_STINT_CHANGES_DATES_PREFIX}:${memberId}`
+    const value = `${organizationId}|${date}`
 
-    // 2. If the date is new, update the set and the queue
-    if (!dates.includes(date)) {
-      dates.push(date)
-      dates.sort()
+    await this.redisClient
+      .multi()
+      .sAdd(datesKey, value)
+      .sAdd(MEMBER_ORG_STINT_CHANGES_QUEUE, memberId)
+      .exec()
 
-      await Promise.all([
-        // Update the specific org field in the hash
-        this.redisClient.hSet(key, organizationId, JSON.stringify(dates)),
-        // Ensure the member is in the processing queue
-        this.redisClient.sAdd(MEMBER_ORG_STINT_CHANGES_QUEUE, memberId),
-      ])
-
-      this.log.debug(
-        { memberId, organizationId, date, count: dates.length },
-        'Buffered activity date and queued member.',
-      )
-    }
+    this.log.debug({ memberId, organizationId, date }, 'Buffered activity date and queued member.')
   }
 }

services/libs/common_services/src/services/common.member.service.ts

@@ -248,13 +248,12 @@ export class CommonMemberService extends LoggerBase {
         highestPrioritySourceExperiences.map((e) => e.organizationId),
       )
 
-      if (memberCounts[0].memberCount > memberCounts[1].memberCount) {
+      // memberCounts is sorted desc by memberCount — pick the winner if it's strictly highest
+      if (memberCounts?.length >= 2 && memberCounts[0].memberCount > memberCounts[1].memberCount) {
         return memberCounts[0].organizationId
-      } else if (memberCounts[0].memberCount < memberCounts[1].memberCount) {
-        return memberCounts[1].organizationId
       }
 
-      // if there's a draw in the member count, use the one with the longer period
+      // tie or no data — fall back to longest date range
       return getLongestDateRange(highestPrioritySourceExperiences).organizationId
     }
 

services/libs/redis/src/cache.ts

@@ -1,7 +1,7 @@
 import { Logger, LoggerBase } from '@crowd/logging'
 import { ICache } from '@crowd/types'
 
-import { RedisClient } from './types'
+import type { RedisClient } from './types'
 
 export class RedisCache extends LoggerBase implements ICache {
   private readonly prefixer: (key: string) => string
@@ -202,6 +202,42 @@ return cjson.encode(results)`
     return map
   }
 
+  /**
+   * Atomically removes members from a set and cleans up the tracking queue
+   * if the set is now empty. Prevents race conditions between SCARD and SREM.
+   *
+   * @returns 1 if the queue entry was removed, 0 otherwise.
+   */
+  public static async ackSetMembers(
+    client: RedisClient,
+    setKey: string,
+    queueKey: string,
+    queueMember: string,
+    members: string[],
+  ): Promise<number> {
+    // Guard clause: Redis errors if you call SREM with no members
+    if (members.length === 0) return 0
+
+    const script = `
+      local chunkSize = 500
+      for i = 2, #ARGV, chunkSize do
+        redis.call('SREM', KEYS[1], unpack(ARGV, i, math.min(i + chunkSize - 1, #ARGV)))
+      end
+
+      if redis.call('SCARD', KEYS[1]) == 0 then
+        return redis.call('SREM', KEYS[2], ARGV[1])
+      end
+      return 0
+    `
+
+    const result = await client.eval(script, {
+      keys: [setKey, queueKey],
+      arguments: [queueMember, ...members],
+    })
+
+    return Number(result)
+  }
+
   public async setIfNotExistsOrGet(
     key: string,
     value: string,