fix misattributed activities based on username + memberId

services/apps/script_executor_worker/src/activities.ts

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

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

@@ -0,0 +1,64 @@
+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(
+  activityIds: string[],
+  correctMemberId: string,
+) {
+  try {
+    const activityRepo = new ActivityRepository(
+      svc.postgres.writer.connection(),
+      svc.log,
+      svc.questdbSQL,
+    )
+    await activityRepo.batchUpdateActivitiesWithWrongMember(activityIds, correctMemberId)
+  } catch (err) {
+    throw new Error(err)
+  }
+}

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,10 @@ export interface IDissectMemberArgs {
   forceSplitAllIdentities?: boolean
 }
 
-export interface IFixOrgIdentitiesWithWrongUrlsArgs {
+export interface tenantIdAndTestRun {
   tenantId: string
   testRun?: boolean
 }
+
+export type IFixMisattributedActivitiesArgs = tenantIdAndTestRun
+export type IFixOrgIdentitiesWithWrongUrlsArgs = tenantIdAndTestRun

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,43 @@
+import { continueAsNew, proxyActivities } from '@temporalio/workflow'
+
+import * as activities from '../activities'
+import { IFixMisattributedActivitiesArgs } from '../types'
+
+const activity = proxyActivities<typeof activities>({
+  startToCloseTimeout: '10 minute',
+  retry: { maximumAttempts: 3 },
+})
+
+export async function fixMisattributedActivities(
+  args: IFixMisattributedActivitiesArgs,
+): Promise<void> {
+  const PROCESS_ACTIVITIES_PER_RUN = args.testRun ? 10 : 1000
+
+  if (args.testRun) {
+    console.log(`Running in test mode with limit 10!`)
+  }
+
+  const tenantId = args.tenantId
+
+  const activitiesWithWrongMember = await activity.findActivitiesWithWrongMembers(
+    tenantId,
+    PROCESS_ACTIVITIES_PER_RUN,
+  )
+
+  if (!activitiesWithWrongMember.length) {
+    console.log(`No activities found with misattributed members!`)
+    return
+  }
+
+  // Process each group of activities
+  for (const group of activitiesWithWrongMember) {
+    await activity.batchUpdateActivitiesWithWrongMember(group.activityIds, group.correctMemberId)
+  }
+
+  if (!args.testRun) {
+    await continueAsNew<typeof fixMisattributedActivities>({
+      testRun: args.testRun,
+      tenantId: args.tenantId,
+    })
+  }
+}

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

@@ -0,0 +1,107 @@
+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<{ correctMemberId: string; activityIds: string[] }[]> {
+    try {
+      return await this.connection.query(
+        `
+        SELECT 
+          mi."memberId" as "correctMemberId",
+          array_agg(a.id) as "activityIds"
+        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 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(
+    activityIds: string[],
+    correctMemberId: string,
+  ): Promise<void> {
+    try {
+      // Batch update activities in PostgreSQL
+      await this.connection.none(
+        `
+        UPDATE activities
+        SET "memberId" = $(correctMemberId)
+        WHERE id = ANY($(activityIds))
+        `,
+        {
+          correctMemberId,
+          activityIds,
+        },
+      )
+
+      // Batch update activities in QuestDB
+      await updateActivities(
+        this.questdbSQL,
+        async () => ({ memberId: correctMemberId }),
+        'id IN ($(activityIds:csv))',
+        {
+          activityIds: activityIds.join(','),
+        },
+      )
+    } catch (err) {
+      this.log.error('Error while batch updating activities!', err)
+      throw new Error(err)
+    }
+  }
+}

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

@@ -1,6 +1,6 @@
 import { DbConnection, DbTransaction } from '@crowd/database'
 import { Logger } from '@crowd/logging'
-import { IMember } from '@crowd/types'
+import { IMember, IMemberIdentity } from '@crowd/types'
 
 import { IFindMemberIdentitiesGroupedByPlatformResult, ISimilarMember } from './types'
 
@@ -159,6 +159,36 @@ class MemberRepository {
 
     return member
   }
+
+  async findMemberIdentity(
+    username: string,
+    platform: string,
+    tenantId: string,
+    type = 'username',
+  ): Promise<IMemberIdentity | null> {
+    let memberIdentity: IMemberIdentity
+
+    try {
+      memberIdentity = await this.connection.oneOrNone(
+        `
+        select * from "memberIdentities" where value = $(username) and platform = $(platform) 
+        and type = $(type) and verified = true and "tenantId" = $(tenantId)
+      `,
+        {
+          username,
+          platform,
+          type,
+          tenantId,
+        },
+      )
+    } catch (err) {
+      this.log.error('Error while finding member identity!', err)
+
+      throw new Error(err)
+    }
+
+    return memberIdentity
+  }
 }
 
 export default MemberRepository