add temp table to track processed stuff

Author: skwowet

services/apps/script_executor_worker/src/activities/fix-deleted-member-org-affilations/index.ts

@@ -6,6 +6,7 @@ import {
 import { runMemberAffiliationsUpdate } from '@crowd/data-access-layer/src/old/apps/profiles_worker'
 import ActivityRepository from '@crowd/data-access-layer/src/old/apps/script_executor_worker/activity.repo'
 import MemberRepository from '@crowd/data-access-layer/src/old/apps/script_executor_worker/member.repo'
+import TempRepository from '@crowd/data-access-layer/src/old/apps/script_executor_worker/temp.repo'
 
 import { svc } from '../../main'
 
@@ -50,3 +51,11 @@ export async function calculateMemberAffiliations(memberId: string): Promise<voi
 export async function addOrgIdToRedisCache(orgId: string): Promise<void> {
   await svc.redis.sAdd('organizationIdsForAggComputation', orgId)
 }
+
+export async function markMemberOrgAffiliationAsProcessed(
+  memberId: string,
+  organizationId: string,
+): Promise<void> {
+  const repo = new TempRepository(svc.postgres.writer.connection())
+  await repo.markMemberOrgAffiliationAsProcessed(memberId, organizationId)
+}

services/apps/script_executor_worker/src/workflows/fixDeletedMemberOrgAffilations.ts

@@ -35,6 +35,10 @@ export async function fixDeletedMemberOrgAffilations(
     // 2. Check if they have activity in questDb
     const activityCount = await activity.getActivities(memberId, organizationId)
 
+    console.log(
+      `Found ${activityCount} activities for member ${memberId} and org ${organizationId}`,
+    )
+
     // 2.1 If no activities found, we need to get and insert them
     if (activityCount === 0) {
       console.log(
@@ -52,8 +56,11 @@ export async function fixDeletedMemberOrgAffilations(
     await syncActivity.syncMembersBatch([memberId], true)
 
     // 5. Add organizationId to redisCache for sync
-    console.log(`Adding org ${organizationId} to redisCache for sync`)
+    console.log(`Adding org ${organizationId} to redisCache`)
     await activity.addOrgIdToRedisCache(organizationId)
+
+    // 6. Mark member-org affiliation as processed
+    await activity.markMemberOrgAffiliationAsProcessed(memberId, organizationId)
   }
 
   if (args.testRun) {

services/libs/data-access-layer/src/old/apps/script_executor_worker/member.repo.ts

@@ -179,11 +179,16 @@ class MemberRepository {
     // Finds activities where:
     // 1. Member had a work experience (now deleted)
     // 2. Member has NO active work experience at same org
+    // 3. Member-Org pair hasn't been processed yet
     const query = `
       select distinct a."memberId", a."organizationId"
       from activities a
+      left join "processedMemberOrgAffiliations" p on
+        p."memberId" = a."memberId"
+        and p."organizationId" = a."organizationId"
       where a."updatedAt" < $(date)
       and a."organizationId" is not null
+      and p."memberId" is null
       and exists (
           select 1
           from "memberOrganizations" mo

services/libs/data-access-layer/src/old/apps/script_executor_worker/temp.repo.ts

@@ -0,0 +1,20 @@
+import { DbConnection, DbTransaction } from '@crowd/database'
+
+class TempRepository {
+  constructor(private readonly connection: DbConnection | DbTransaction) {}
+
+  async markMemberOrgAffiliationAsProcessed(
+    memberId: string,
+    organizationId: string,
+  ): Promise<void> {
+    await this.connection.none(
+      `insert into "processedMemberOrgAffiliations" ("memberId", "organizationId") values ($(memberId), $(organizationId))`,
+      {
+        memberId,
+        organizationId,
+      },
+    )
+  }
+}
+
+export default TempRepository