fix misattributed activities based on username + memberId

services/apps/script_executor_worker/package.json

@@ -21,6 +21,7 @@
     "@crowd/types": "workspace:*",
     "@temporalio/workflow": "~1.11.1",
     "axios": "^1.6.8",
+    "csv-parse": "^5.5.6",
     "moment": "~2.29.4",
     "tsx": "^4.7.1",
     "typescript": "^5.6.3"

services/apps/script_executor_worker/src/activities.ts

@@ -10,6 +10,13 @@ import {
   findMemberIdentitiesGroupedByPlatform,
   findMemberMergeActions,
 } from './activities/dissect-member'
+import {
+  batchUpdateActivitiesWithWrongMember,
+  findActivitiesWithWrongMembers,
+  findMemberIdentity,
+  updateActivityWithWrongMember,
+  updateMemberActivitiesUpdatedAt,
+} from './activities/fix-misattributed-activities'
 import {
   deleteOrganizationIdentity,
   findOrganizationIdentity,
@@ -38,4 +45,9 @@ export {
   updateOrganizationIdentity,
   deleteOrganizationIdentity,
   isLfxMember,
+  findActivitiesWithWrongMembers,
+  findMemberIdentity,
+  updateActivityWithWrongMember,
+  batchUpdateActivitiesWithWrongMember,
+  updateMemberActivitiesUpdatedAt,
 }

services/apps/script_executor_worker/src/activities/fix-misattributed-activities/index.ts

@@ -0,0 +1,96 @@
+import { parse } from 'csv-parse/sync'
+import * as fs from 'fs'
+import * as path from 'path'
+
+import { ActivityRepository } from '@crowd/data-access-layer/src/old/apps/script_executor_worker/activities.repo'
+import MemberRepository from '@crowd/data-access-layer/src/old/apps/script_executor_worker/member.repo'
+import { IMemberIdentity } from '@crowd/types'
+
+import { svc } from '../../main'
+
+// export async function findActivitiesWithWrongMembers(tenantId: string, limit: number) {
+//   let activitiesWithWrongMember = []
+
+//   try {
+//     const activityRepo = new ActivityRepository(
+//       svc.postgres.reader.connection(),
+//       svc.log,
+//       svc.questdbSQL,
+//     )
+//     activitiesWithWrongMember = await activityRepo.getActivitiesWithWrongMembers(tenantId, limit)
+//   } catch (err) {
+//     throw new Error(err)
+//   }
+
+//   return activitiesWithWrongMember
+// }
+
+export async function findMemberIdentity(username: string, platform: string, tenantId: string) {
+  let memberIdentity: IMemberIdentity
+
+  try {
+    const memberRepo = new MemberRepository(svc.postgres.reader.connection(), svc.log)
+    memberIdentity = await memberRepo.findMemberIdentity(username, platform, tenantId)
+  } catch (err) {
+    throw new Error(err)
+  }
+
+  return memberIdentity
+}
+
+export async function updateActivityWithWrongMember(activityId: string, correctMemberId: string) {
+  try {
+    const activityRepo = new ActivityRepository(
+      svc.postgres.writer.connection(),
+      svc.log,
+      svc.questdbSQL,
+    )
+    await activityRepo.updateActivityWithWrongMember(activityId, correctMemberId)
+  } catch (err) {
+    throw new Error(err)
+  }
+}
+
+export async function batchUpdateActivitiesWithWrongMember(
+  wrongMemberId: string,
+  correctMemberId: string,
+) {
+  try {
+    const activityRepo = new ActivityRepository(
+      svc.postgres.writer.connection(),
+      svc.log,
+      svc.questdbSQL,
+    )
+    await activityRepo.batchUpdateActivitiesWithWrongMember(wrongMemberId, correctMemberId)
+  } catch (err) {
+    throw new Error(err)
+  }
+}
+
+export async function updateMemberActivitiesUpdatedAt(memberId: string) {
+  try {
+    const activityRepo = new ActivityRepository(
+      svc.postgres.writer.connection(),
+      svc.log,
+      svc.questdbSQL,
+    )
+    await activityRepo.updateMemberActivitiesUpdatedAt(memberId)
+  } catch (err) {
+    throw new Error(err)
+  }
+}
+
+export async function findActivitiesWithWrongMembers(): Promise<
+  Array<{
+    wrongMemberId: string
+    correctMemberId: string
+    activitiesCount: number
+  }>
+> {
+  const csvFilePath = path.join(process.cwd(), 'misattributed_activities.csv')
+  const fileContent = fs.readFileSync(csvFilePath, 'utf-8')
+  return parse(fileContent, {
+    columns: true,
+    skip_empty_lines: true,
+  })
+}

services/apps/script_executor_worker/src/main.ts

@@ -10,7 +10,7 @@ const config: Config = {
     enabled: true,
   },
   questdb: {
-    enabled: false,
+    enabled: true,
   },
   redis: {
     enabled: true,

services/apps/script_executor_worker/src/types.ts

@@ -22,7 +22,14 @@ export interface IDissectMemberArgs {
   forceSplitAllIdentities?: boolean
 }
 
-export interface IFixOrgIdentitiesWithWrongUrlsArgs {
+export interface tenantIdAndTestRun {
   tenantId: string
   testRun?: boolean
 }
+
+export type IFixOrgIdentitiesWithWrongUrlsArgs = tenantIdAndTestRun
+
+export interface IFixMisattributedActivitiesArgs {
+  startIndex?: number
+  testRun?: boolean
+}

services/apps/script_executor_worker/src/workflows.ts

@@ -1,11 +1,13 @@
 import { dissectMember } from './workflows/dissectMember'
 import { findAndMergeMembersWithSamePlatformIdentitiesDifferentCapitalization } from './workflows/findAndMergeMembersWithSamePlatformIdentitiesDifferentCapitalization'
 import { findAndMergeMembersWithSameVerifiedEmailsInDifferentPlatforms } from './workflows/findAndMergeMembersWithSameVerifiedEmailsInDifferentPlatforms'
+import { fixMisattributedActivities } from './workflows/fixMisattributedActivities'
 import { fixOrgIdentitiesWithWrongUrls } from './workflows/fixOrgIdentitiesWithWrongUrls'
 
 export {
   findAndMergeMembersWithSameVerifiedEmailsInDifferentPlatforms,
   findAndMergeMembersWithSamePlatformIdentitiesDifferentCapitalization,
   dissectMember,
   fixOrgIdentitiesWithWrongUrls,
+  fixMisattributedActivities,
 }

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

@@ -0,0 +1,58 @@
+import { proxyActivities } from '@temporalio/workflow'
+
+import * as activities from '../activities'
+import { IFixMisattributedActivitiesArgs } from '../types'
+
+const activity = proxyActivities<typeof activities>({
+  startToCloseTimeout: '60 minutes',
+})
+
+export async function fixMisattributedActivities(
+  args: IFixMisattributedActivitiesArgs,
+): Promise<void> {
+  // Read CSV file
+  const records = await activity.findActivitiesWithWrongMembers()
+
+  if (!records.length) {
+    console.log(`No activities found in the CSV file!`)
+    return
+  }
+
+  const startIndex = Number(args.startIndex)
+
+  if (!startIndex) {
+    console.log('something wrong with startIndex')
+    return
+  }
+
+  console.log(`Starting at record ${startIndex}`)
+
+  // Skip records that were already processed
+  const remainingRecords = records.slice(startIndex)
+
+  if (!remainingRecords.length) {
+    console.log(`No remaining records to process after skipping ${startIndex} records!`)
+    return
+  }
+
+  let processedMemberCount = 0
+  const totalRecords = remainingRecords.length
+
+  // Process each record from CSV
+  for (const record of remainingRecords) {
+    console.log(`Updating ${record.correctMemberId} member activities updatedAt`)
+
+    await activity.updateMemberActivitiesUpdatedAt(record.correctMemberId)
+
+    processedMemberCount++
+
+    if (args.testRun && processedMemberCount >= 10) {
+      console.log('Test run complete!')
+      break
+    }
+
+    console.log(`Processed ${processedMemberCount}/${totalRecords} members in the CSV file!`)
+  }
+
+  console.log('Completed processing all members!')
+}

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

@@ -0,0 +1,106 @@
+import { DbConnOrTx, DbConnection, DbTransaction } from '@crowd/database'
+import { Logger } from '@crowd/logging'
+
+import { updateActivities } from '../../../activities/update'
+
+export class ActivityRepository {
+  constructor(
+    private readonly connection: DbConnection | DbTransaction,
+    private readonly log: Logger,
+    private readonly questdbSQL: DbConnOrTx,
+  ) {}
+
+  async getActivitiesWithWrongMembers(
+    tenantId: string,
+    limit = 100,
+  ): Promise<{ wrongMemberId: string; correctMemberId: string }[]> {
+    try {
+      return await this.connection.query(
+        `
+        SELECT 
+          a."memberId" as "wrongMemberId",
+          mi."memberId" as "correctMemberId"
+        FROM activities a
+        JOIN "memberIdentities" mi ON a.username = mi.value
+          AND a.platform = mi.platform 
+          AND mi.type = 'username'
+          AND mi."verified" = true
+          AND a."tenantId" = mi."tenantId"
+        WHERE a."memberId" <> mi."memberId"
+          AND a."tenantId" = $(tenantId)
+        GROUP BY a."memberId", mi."memberId"
+        LIMIT $(limit)
+        `,
+        {
+          tenantId,
+          limit,
+        },
+      )
+    } catch (err) {
+      this.log.error('Error while finding activities!', err)
+      throw new Error(err)
+    }
+  }
+
+  async updateActivityWithWrongMember(activityId: string, correctMemberId: string): Promise<void> {
+    try {
+      // Update the activity in pgsql to persist progress
+      await this.connection.none(
+        `
+        UPDATE activities
+        SET "memberId" = $(correctMemberId)
+        WHERE id = $(activityId)
+        `,
+        {
+          correctMemberId,
+          activityId,
+        },
+      )
+
+      // Update the activity in QuestDB
+      await updateActivities(
+        this.questdbSQL,
+        async () => ({ memberId: correctMemberId }),
+        'id = $(activityId)',
+        {
+          activityId,
+        },
+      )
+    } catch (err) {
+      this.log.error('Error while updating activities!', err)
+      throw new Error(err)
+    }
+  }
+
+  async batchUpdateActivitiesWithWrongMember(
+    wrongMemberId: string,
+    correctMemberId: string,
+  ): Promise<void> {
+    try {
+      await this.questdbSQL.none(
+        'UPDATE "activities" SET "memberId" = $(correctMemberId) WHERE "memberId" = $(wrongMemberId)',
+        {
+          wrongMemberId,
+          correctMemberId,
+        },
+      )
+    } catch (err) {
+      this.log.error('Error while batch updating activities!', err)
+      throw new Error(err)
+    }
+  }
+
+  async updateMemberActivitiesUpdatedAt(memberId: string): Promise<void> {
+    try {
+      await this.questdbSQL.none(
+        'UPDATE "activities" SET "updatedAt" = now() WHERE "memberId" = $(memberId)',
+        {
+          memberId,
+        },
+      )
+    } catch (err) {
+      this.log.error('Error while updating activities!', err)
+      throw new Error(err)
+    }
+  }
+}