fix: resolve pr comments from ai bots

Signed-off-by: Yeganathan S <63534555+skwowet@users.noreply.github.com>

Author: skwowet

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

@@ -1,18 +1,16 @@
 import CronTime from 'cron-time-generator'
-import { 
-  inferMemberOrganizationStintChanges, 
-  MEMBER_ORG_STINT_CHANGES_DATES_PREFIX, 
-  MEMBER_ORG_STINT_CHANGES_QUEUE 
-} from '@crowd/common_services'
+
 import {
-  createMemberOrganization,
-  fetchMemberOrganizationsBySource,
-  updateMemberOrganization,
-} from '@crowd/data-access-layer'
+  MEMBER_ORG_STINT_CHANGES_DATES_PREFIX,
+  MEMBER_ORG_STINT_CHANGES_QUEUE,
+  inferMemberOrganizationStintChanges,
+} from '@crowd/common_services'
+import { fetchMemberOrganizationsBySource } from '@crowd/data-access-layer'
 import { WRITE_DB_CONFIG, getDbConnection } from '@crowd/data-access-layer/src/database'
-import { pgpQx, QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor'
-import { REDIS_CONFIG, RedisClient, getRedisClient } from '@crowd/redis'
-import { MemberOrgDate, MemberOrgStintChange, OrganizationSource } from '@crowd/types'
+import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor'
+import { REDIS_CONFIG, getRedisClient } from '@crowd/redis'
+import { OrganizationSource } from '@crowd/types'
+
 import { IJobDefinition } from '../types'
 
 const job: IJobDefinition = {
@@ -24,106 +22,82 @@ const job: IJobDefinition = {
     const db = await getDbConnection(WRITE_DB_CONFIG(), 2, 0)
     const qx = pgpQx(db)
 
-    // 1. Fetch a batch of work triggers (Member IDs)
-    const memberIds = await redis.sPop(MEMBER_ORG_STINT_CHANGES_QUEUE, 500)
+    // 1. Get a batch of work
+    const memberIds = await redis.sRandMember(MEMBER_ORG_STINT_CHANGES_QUEUE, 500)
     if (!memberIds?.length) return
 
     ctx.log.info({ count: memberIds.length }, 'Processing pending members.')
     const stats = { processed: 0, inserts: 0, updates: 0 }
 
     for (const memberId of memberIds) {
       try {
-        // 2. Get the activity dates for this member
-        const activityDates = await popMemberOrganizationActivityDates(redis, memberId)
-        
-        // If the member has no activity dates, move to the next member
-        if (activityDates.length === 0) continue
-
-        // 3. Sync with existing database state
-        const existingOrgs = await fetchMemberOrganizationsBySource(
-          qx,
-          memberId,
-          OrganizationSource.EMAIL_DOMAIN,
-        )
-
-        // 4. Calculate required stint changes
-        const stintChanges = inferMemberOrganizationStintChanges(
-          memberId,
-          existingOrgs,
-          activityDates,
-        )
-
-        if (stintChanges.length === 0) continue
-
-        const counts = {
-          inserts: stintChanges.filter((c) => c.type === 'insert').length,
-          updates: stintChanges.filter((c) => c.type === 'update').length,
+        const datesKey = `${MEMBER_ORG_STINT_CHANGES_DATES_PREFIX}:${memberId}`
+        const hash = await redis.hGetAll(datesKey)
+
+        // If no data, just remove from queue and move on
+        if (!hash || Object.keys(hash).length === 0) {
+          await redis.sRem(MEMBER_ORG_STINT_CHANGES_QUEUE, memberId)
+          continue
         }
 
-        ctx.log.info({ memberId, ...counts }, 'Stint changes identified.')
+        // 2. Parse Redis data into domain objects
+        const { activityDates, orgIds } = parseMemberActivityHash(hash)
 
-        ctx.log.debug(
-          { memberId, activityDates, existingOrgs, stintChanges },
-          'Stint inference trace.',
-        )
+        if (activityDates.length > 0) {
+          // 3. Compare with DB and calculate delta
+          const existingOrgs = await fetchMemberOrganizationsBySource(
+            qx,
+            memberId,
+            OrganizationSource.EMAIL_DOMAIN,
+          )
 
-        // @todo: Enable writes after dry-run validation
-        // await applyStintChanges(qx, stintChanges) 
+          const changes = inferMemberOrganizationStintChanges(memberId, existingOrgs, activityDates)
+
+          if (changes.length > 0) {
+            ctx.log.info({ memberId, count: changes.length }, 'Stint changes identified.')
+            stats.inserts += changes.filter((c) => c.type === 'insert').length
+            stats.updates += changes.filter((c) => c.type === 'update').length
+          }
+        }
+
+        // 4. Cleanup: Remove only the fields we actually read
+        await redis
+          .multi()
+          .hDel(datesKey, ...orgIds)
+          .sRem(MEMBER_ORG_STINT_CHANGES_QUEUE, memberId)
+          .exec()
 
         stats.processed++
-        stats.inserts += counts.inserts
-        stats.updates += counts.updates
       } catch (err) {
         ctx.log.error(err, { memberId }, 'Failed to process member stint inference.')
       }
     }
 
-    ctx.log.info(stats, 'Batch inference complete.')
+    ctx.log.info(stats, 'Batch complete.')
   },
 }
 
-async function popMemberOrganizationActivityDates(
-  redis: RedisClient,
-  memberId: string,
-): Promise<MemberOrgDate[]> {
-  const key = `${MEMBER_ORG_STINT_CHANGES_DATES_PREFIX}:${memberId}`
-  
-  // hGetAll + del in a multi block makes the "Pop" atomic for the entire Hash
-  const [hash] = (await redis.multi().hGetAll(key).del(key).exec()) as [Record<string, string> | null, number]
-
-  if (!hash || Object.keys(hash).length === 0) return []
-
-  return Object.entries(hash)
-    .flatMap(([organizationId, datesJson]) =>
-      (JSON.parse(datesJson) as string[]).map((date) => ({ organizationId, date })),
-    )
+/**
+ * Parses the Redis hash into a clean, typed list of 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 []
+      }
+    })
     .sort((a, b) => a.date.localeCompare(b.date))
-}
-
-async function applyStintChanges(
-  qx: QueryExecutor,
-  stintChanges: MemberOrgStintChange[],
-): Promise<void> {
-  for (const change of stintChanges) {
-    if (change.type === 'insert') {
-      await createMemberOrganization(qx, change.memberId, {
-        organizationId: change.organizationId,
-        dateStart: change.dateStart,
-        dateEnd: change.dateEnd,
-        source: OrganizationSource.EMAIL_DOMAIN,
-      })
-      continue
-    }
-
-    if (!change.id) {
-      throw new Error('Missing id for update stint change.')
-    }
 
-    await updateMemberOrganization(qx, change.memberId, change.id, {
-      dateStart: change.dateStart,
-      dateEnd: change.dateEnd,
-    })
-  }
+  return { activityDates, orgIds }
 }
 
-export default job
+export default job

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

@@ -1076,29 +1076,42 @@ export default class MemberService extends LoggerBase {
     organizationId: string,
     activityTimestamp: string,
   ): Promise<void> {
-    // 1. Normalize the timestamp to a simple YYYY-MM-DD string
-    const date = new Date(activityTimestamp).toISOString().slice(0, 10)
+    const date = new Date(activityTimestamp).toISOString().split('T')[0]
     const key = `${MEMBER_ORG_STINT_CHANGES_DATES_PREFIX}:${memberId}`
 
-    // 2. Fetch and de-duplicate the date within the organization's buffer
+    // 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('Corrupt dates buffer value detected during buffering.')
+        return []
+      }
+    }
+
+    // 1. Fetch current dates for this specific organization
     const existing = await this.redisClient.hGet(key, organizationId)
-    const dates: string[] = existing ? JSON.parse(existing) : []
+    const dates: string[] = parseExistingDates(existing)
 
-    if (dates.includes(date)) {
-      this.log.trace({ memberId, organizationId, date }, 'Date already buffered, skipping.')
-    } else {
+    // 2. If the date is new, update the set and the queue
+    if (!dates.includes(date)) {
       dates.push(date)
       dates.sort()
 
-      // 3. Update the buffer with the new sorted date list
-      await this.redisClient.hSet(key, organizationId, JSON.stringify(dates))
+      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, totalDates: dates.length },
-        'Buffered activity date for stint inference.',
+        { memberId, organizationId, date, count: dates.length },
+        'Buffered activity date and queued member.',
       )
     }
-
-    // 4. add the member to the inference queue
-    await this.redisClient.sAdd(MEMBER_ORG_STINT_CHANGES_QUEUE, memberId)
   }
 }

services/libs/common_services/src/services/member-organization.ts

@@ -1,4 +1,3 @@
-import { RedisClient } from '@crowd/redis'
 import {
   IMemberOrganization,
   MemberOrgDate,
@@ -130,6 +129,8 @@ interface Stint {
   isNew: boolean
 }
 
+type DatedStint = Stint & { dateStart: string; dateEnd: string }
+
 /**
  * Core logic to determine if activity dates should expand existing stints or create new ones
  */
@@ -163,7 +164,9 @@ export function inferMemberOrganizationStintChanges(
       continue
 
     // 3. Fill undated placeholder only when no dated stint exists yet (Rule 2)
-    const dated = orgStints.filter((s) => s.dateStart && s.dateEnd)
+    const dated = orgStints.filter(
+      (s): s is DatedStint => s.dateStart !== null && s.dateEnd !== null,
+    )
     const placeholder = orgStints.find((s) => !s.dateStart && !s.dateEnd)
     if (placeholder && dated.length === 0) {
       placeholder.dateStart = placeholder.dateEnd = targetDate
@@ -172,12 +175,12 @@ export function inferMemberOrganizationStintChanges(
     }
 
     // 4. Find the closest neighbor stint to see if expansion is possible
-    let neighbor: Stint | null = null
+    let neighbor: DatedStint | null = null
     let minGap = Infinity
 
     for (const s of dated) {
       const gap =
-        targetDate > s.dateEnd! ? diff(s.dateEnd!, targetDate) : diff(targetDate, s.dateStart!)
+        targetDate > s.dateEnd ? diff(s.dateEnd, targetDate) : diff(targetDate, s.dateStart)
       if (gap < minGap) {
         minGap = gap
         neighbor = s
@@ -198,8 +201,9 @@ export function inferMemberOrganizationStintChanges(
 
     // 5. Determine the gap window between neighbor and targetDate, then check if another org
     // holds a significant (>= 30 day) dated stint that overlaps that window (multi-stint guard)
-    const gapStart = targetDate > neighbor.dateEnd! ? neighbor.dateEnd! : targetDate
-    const gapEnd = targetDate > neighbor.dateEnd! ? targetDate : neighbor.dateStart!
+    const isForward = targetDate > neighbor.dateEnd
+    const gapStart = isForward ? neighbor.dateEnd : targetDate
+    const gapEnd = isForward ? targetDate : neighbor.dateStart
     const hasConflict = stints.some(
       (s) =>
         s.organizationId !== organizationId &&
@@ -210,8 +214,6 @@ export function inferMemberOrganizationStintChanges(
         diff(s.dateStart, s.dateEnd) >= 30,
     )
 
-    const isForward = targetDate > neighbor.dateEnd!
-
     if (hasConflict) {
       // 6a. Another org owns the gap — start a fresh stint rather than bridging
       stints.push({