refactor: use activityRelations for affiliations, merges, and others

backend/src/bin/jobs/syncActivities.ts

@@ -6,7 +6,6 @@ import { IDbActivityCreateData } from '@crowd/data-access-layer/src/old/apps/dat
 import ActivityRepository from '@crowd/data-access-layer/src/old/apps/data_sink_worker/repo/activity.repo'
 import { QueryExecutor, formatQuery, pgpQx } from '@crowd/data-access-layer/src/queryExecutor'
 import { Logger, logExecutionTimeV2, timer } from '@crowd/logging'
-import { getClientSQL } from '@crowd/questdb'
 import { PlatformType } from '@crowd/types'
 
 import { DB_CONFIG } from '@/conf'
@@ -16,16 +15,18 @@ import { retryBackoff } from '../../utils/backoff'
 
 async function decideUpdatedAt(pgQx: QueryExecutor, maxUpdatedAt?: string): Promise<string> {
   if (!maxUpdatedAt) {
-    const result = await pgQx.selectOne('SELECT MAX("updatedAt") AS "maxUpdatedAt" FROM activities')
+    const result = await pgQx.selectOne(
+      'SELECT MAX("updatedAt") AS "maxUpdatedAt" FROM "activityRelations"',
+    )
     return result?.maxUpdatedAt
   }
 
   return maxUpdatedAt
 }
 
-async function getTotalActivities(qdbQx: QueryExecutor, whereClause: string): Promise<number> {
-  const { totalActivities } = await qdbQx.selectOne(
-    `SELECT COUNT(1) AS "totalActivities" FROM activities WHERE ${whereClause}`,
+async function getTotalActivities(pgQx: QueryExecutor, whereClause: string): Promise<number> {
+  const { totalActivities } = await pgQx.selectOne(
+    `SELECT COUNT(1) AS "totalActivities" FROM "activityRelations" WHERE ${whereClause}`,
   )
   return totalActivities
 }
@@ -78,7 +79,6 @@ async function syncActivitiesBatch({
 export async function syncActivities(logger: Logger, maxUpdatedAt?: string) {
   logger.info(`Syncing activities from ${maxUpdatedAt}`)
 
-  const qdb = await getClientSQL()
   const db = await getDbConnection({
     host: DB_CONFIG.writeHost,
     port: DB_CONFIG.port,
@@ -88,7 +88,6 @@ export async function syncActivities(logger: Logger, maxUpdatedAt?: string) {
   })
 
   const pgQx = pgpQx(db)
-  const qdbQx = pgpQx(qdb)
   const activityRepo = new ActivityRepository(new DbStore(logger, db, undefined, true), logger)
 
   let updatedAt = await logExecutionTimeV2(
@@ -100,7 +99,7 @@ export async function syncActivities(logger: Logger, maxUpdatedAt?: string) {
   const whereClause = createWhereClause(updatedAt)
 
   const totalActivities = await logExecutionTimeV2(
-    () => getTotalActivities(qdbQx, whereClause),
+    () => getTotalActivities(pgQx, whereClause),
     logger,
     'get total activities',
   )
@@ -114,10 +113,10 @@ export async function syncActivities(logger: Logger, maxUpdatedAt?: string) {
       // eslint-disable-next-line @typescript-eslint/no-loop-func
       () =>
         retryBackoff(() =>
-          qdbQx.select(
+          pgQx.select(
             `
             SELECT *
-            FROM activities
+            FROM "activityRelations"
             WHERE "updatedAt" > $(updatedAt)
             ORDER BY "updatedAt"
             LIMIT 1000;
@@ -126,7 +125,7 @@ export async function syncActivities(logger: Logger, maxUpdatedAt?: string) {
           ),
         ),
       logger,
-      `getting activities with updatedAt > ${updatedAt}`,
+      `getting activityRelations with updatedAt > ${updatedAt}`,
     )
 
     if (result.length === 0) {

backend/src/database/migrations/V1752141258__add_affiliations_read_index.sql

@@ -0,0 +1,2 @@
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "ix_activityRelations_memberId_segmentId_timestamp"
+ON "activityRelations" ("memberId", "segmentId", "timestamp");

backend/src/database/repositories/memberRepository.ts

@@ -17,9 +17,9 @@ import {
   groupBy,
 } from '@crowd/common'
 import {
-  countMembersWithActivities,
   getActiveMembers,
   getLastActivitiesForMembers,
+  queryActivityRelations,
   setMemberDataToActivities,
 } from '@crowd/data-access-layer'
 import { findManyLfxMemberships } from '@crowd/data-access-layer/src/lfx_memberships'
@@ -1437,7 +1437,9 @@ class MemberRepository {
       segments = [originalSegment]
     }
 
-    const activeMemberResults = await getActiveMembers(options.qdb, {
+    const qx = SequelizeRepository.getQueryExecutor(options)
+
+    const activeMemberResults = await getActiveMembers(qx, {
       timestampFrom: new Date(Date.parse(filter.activityTimestampFrom)).toISOString(),
       timestampTo: new Date(Date.parse(filter.activityTimestampTo)).toISOString(),
       isContribution: filter.activityIsContribution === true ? true : undefined,
@@ -1565,13 +1567,21 @@ class MemberRepository {
   }
 
   static async countMembersPerSegment(options: IRepositoryOptions, segmentIds: string[]) {
-    const countResults = await countMembersWithActivities(options.qdb, {
-      segmentIds,
+    const qx = SequelizeRepository.getQueryExecutor(options)
+    const result = await queryActivityRelations(qx, {
+      filter: {
+        and: [
+          {
+            segmentId: {
+              in: segmentIds,
+            },
+          },
+        ],
+      },
+      countOnly: true,
     })
-    return countResults.reduce((acc, curr: any) => {
-      acc[curr.segmentId] = parseInt(curr.totalCount, 10)
-      return acc
-    }, {})
+
+    return result.count
   }
 
   static async countMembers(options: IRepositoryOptions, segmentIds: string[]) {
@@ -1710,9 +1720,10 @@ class MemberRepository {
       row.activityCount = parseInt(row.activityCount, 10)
     }
 
+    const qx = SequelizeRepository.getQueryExecutor(options)
+
     const memberIds = translatedRows.map((r) => r.id)
     if (memberIds.length > 0) {
-      const qx = SequelizeRepository.getQueryExecutor(options)
       const organizationIds = uniq(
         translatedRows.reduce((acc, r) => {
           acc.push(...r.organizations.map((o) => o.id))
@@ -1729,7 +1740,7 @@ class MemberRepository {
         }
       }
 
-      const lastActivities = await getLastActivitiesForMembers(options.qdb, memberIds, segments)
+      const lastActivities = await getLastActivitiesForMembers(qx, options.qdb, memberIds, segments)
 
       for (const row of translatedRows) {
         const r = row as any
@@ -2114,7 +2125,9 @@ class MemberRepository {
     })
 
     if (memberIds.length > 0) {
-      const lastActivities = await getLastActivitiesForMembers(options.qdb, memberIds, [segmentId])
+      const lastActivities = await getLastActivitiesForMembers(qx, options.qdb, memberIds, [
+        segmentId,
+      ])
 
       rows.forEach((r) => {
         r.lastActivity = lastActivities.find((a) => a.memberId === r.id)

backend/src/database/repositories/organizationRepository.ts

@@ -10,9 +10,9 @@ import {
 } from '@crowd/audit-logs'
 import { Error400, Error404, Error409, PageData, RawQueryParser } from '@crowd/common'
 import {
-  countMembersWithActivities,
   getActiveOrganizations,
   queryActivities,
+  queryActivityRelations,
 } from '@crowd/data-access-layer'
 import { findManyLfxMemberships } from '@crowd/data-access-layer/src/lfx_memberships'
 import {
@@ -1404,7 +1404,9 @@ class OrganizationRepository {
       segments = [originalSegment]
     }
 
-    const activeOrgsResults = await getActiveOrganizations(options.qdb, {
+    const qx = SequelizeRepository.getQueryExecutor(options)
+
+    const activeOrgsResults = await getActiveOrganizations(qx, {
       timestampFrom: new Date(Date.parse(filter.activityTimestampFrom)),
       timestampTo: new Date(Date.parse(filter.activityTimestampTo)),
       platforms: filter.platforms ? filter.platforms : undefined,
@@ -1491,7 +1493,6 @@ class OrganizationRepository {
       options,
     )
 
-    const qx = SequelizeRepository.getQueryExecutor(options)
     const lfxMemberships = await findManyLfxMemberships(qx, {
       organizationIds,
     })
@@ -1876,17 +1877,24 @@ class OrganizationRepository {
     platform: string,
     options: IRepositoryOptions,
   ): Promise<number> {
-    const rows = await countMembersWithActivities(options.qdb, {
-      organizationId,
-      platform,
-    })
-
-    let count = 0
-    rows.forEach((row) => {
-      count += Number(row.count)
+    const qx = SequelizeRepository.getQueryExecutor(options)
+    const rows = await queryActivityRelations(qx, {
+      filter: {
+        and: [
+          {
+            organizationId: {
+              eq: organizationId,
+            },
+            platform: {
+              eq: platform,
+            },
+          },
+        ],
+      },
+      countOnly: true,
     })
 
-    return count
+    return rows.count
   }
 
   static async removeIdentitiesFromOrganization(

backend/src/services/memberService.ts

@@ -27,7 +27,6 @@ import {
 } from '@crowd/data-access-layer/src/members'
 import { findMergeAction } from '@crowd/data-access-layer/src/mergeActions/repo'
 import { QueryExecutor, optionsQx } from '@crowd/data-access-layer/src/queryExecutor'
-// import { getActivityCountOfMemberIdentities } from '@crowd/data-access-layer'
 import { fetchManySegments } from '@crowd/data-access-layer/src/segments'
 import { LoggerBase } from '@crowd/logging'
 import {
@@ -1045,20 +1044,6 @@ export default class MemberService extends LoggerBase {
           )
           member.memberOrganizations = unmergedRoles as IMemberRoleWithOrganization[]
 
-          const secondaryActivityCount = 0
-          const primaryActivityCount = 0
-          // activity count
-          // const secondaryActivityCount = await getActivityCountOfMemberIdentities(
-          //   this.options.qdb,
-          //   member.id,
-          //   secondaryBackup.identities,
-          // )
-          // const primaryActivityCount = await getActivityCountOfMemberIdentities(
-          //   this.options.qdb,
-          //   member.id,
-          //   member.identities,
-          // )
-
           return {
             primary: {
               ...lodash.pick(member, MemberService.MEMBER_MERGE_FIELDS),
@@ -1068,15 +1053,13 @@ export default class MemberService extends LoggerBase {
                 member.memberOrganizations,
               ),
               username: MemberRepository.getUsernameFromIdentities(member.identities),
-              activityCount: primaryActivityCount,
               numberOfOpenSourceContributions: member.contributions?.length || 0,
             },
             secondary: {
               ...secondaryBackup,
               organizations: OrganizationRepository.calculateRenderFriendlyOrganizations(
                 secondaryBackup.memberOrganizations,
               ),
-              activityCount: secondaryActivityCount,
               numberOfOpenSourceContributions: secondaryBackup.contributions?.length || 0,
             },
           }
@@ -1108,21 +1091,6 @@ export default class MemberService extends LoggerBase {
         throw new Error400(this.options.language, 'merge.errors.noIdentities')
       }
 
-      const secondaryActivityCount = 0
-      const primaryActivityCount = 0
-
-      // const secondaryActivityCount = await getActivityCountOfMemberIdentities(
-      //   this.options.qdb,
-      //   member.id,
-      //   secondaryIdentities,
-      // )
-      //
-      // const primaryActivityCount = await getActivityCountOfMemberIdentities(
-      //   this.options.qdb,
-      //   member.id,
-      //   primaryIdentities,
-      // )
-
       const primaryMemberRoles = await MemberOrganizationRepository.findMemberRoles(
         member.id,
         this.options,
@@ -1136,7 +1104,6 @@ export default class MemberService extends LoggerBase {
           organizations:
             OrganizationRepository.calculateRenderFriendlyOrganizations(primaryMemberRoles),
           username: MemberRepository.getUsernameFromIdentities(primaryIdentities),
-          activityCount: primaryActivityCount,
           numberOfOpenSourceContributions: member.contributions?.length || 0,
         },
         secondary: {
@@ -1155,7 +1122,6 @@ export default class MemberService extends LoggerBase {
           contributions: [],
           manuallyCreated: true,
           manuallyChangedFields: [],
-          activityCount: secondaryActivityCount,
           numberOfOpenSourceContributions: 0,
         },
       }

services/apps/cache_worker/src/activities/dashboard-cache/refreshDashboardCache.ts

@@ -4,7 +4,7 @@ import {
   activitiesTimeseries,
   getTimeseriesOfActiveMembers,
   getTimeseriesOfNewMembers,
-  queryActivities,
+  queryActivityRelations,
 } from '@crowd/data-access-layer'
 import { DbStore } from '@crowd/data-access-layer/src/database'
 import ActivityRepository from '@crowd/data-access-layer/src/old/apps/cache_worker/activity.repo'
@@ -15,7 +15,7 @@ import {
   getTimeseriesOfActiveOrganizations,
   getTimeseriesOfNewOrganizations,
 } from '@crowd/data-access-layer/src/organizations'
-import { dbStoreQx } from '@crowd/data-access-layer/src/queryExecutor'
+import { dbStoreQx, pgpQx } from '@crowd/data-access-layer/src/queryExecutor'
 import { RedisCache } from '@crowd/redis'
 import {
   DashboardTimeframe,
@@ -109,7 +109,8 @@ export async function getActivitiesNumber(params: IQueryTimeseriesParams): Promi
       })
     }
 
-    const res = await queryActivities(svc.questdbSQL, {
+    const qx = pgpQx(svc.postgres.reader.connection())
+    const res = await queryActivityRelations(qx, {
       segmentIds: params.segmentIds,
       countOnly: true,
       filter: {