refactor: aggregate calculations for members and organizations (#3041)

Author: skwowet

backend/src/database/migrations/U1746535447__activityRelationsIndexesForCoreAggs.sql

@@ -0,0 +1,2 @@
+drop index concurrently if exists "ix_activityRelations_memberId_segmentId_include";
+drop index concurrently if exists "ix_activityRelations_organizationId_segmentId_include";

backend/src/database/migrations/U1747131692__systemSettingsAndSegmentsAgg.sql

@@ -0,0 +1,10 @@
+alter table "memberSegmentsAgg"
+    drop column if exists "updatedAt";
+
+alter table "organizationSegmentsAgg"
+    drop column if exists "createdAt";
+
+alter table "organizationSegmentsAgg"
+    drop column if exists "updatedAt";
+
+drop table if exists "systemSettings";

backend/src/database/migrations/U1747133060__optimizeIndexedEntitiesIndexes.sql

@@ -0,0 +1 @@
+drop index concurrently if exists "idx_indexed_entities_type_indexed_at_entity_id";

backend/src/database/migrations/V1746535447__activityRelationsIndexesForCoreAggs.sql

@@ -0,0 +1,8 @@
+-- Indexes to optimize queries for member and organization activity core aggregates
+create index concurrently if not exists "ix_activityRelations_memberId_segmentId_include"
+on "activityRelations" ("memberId", "segmentId")
+include ("platform", "activityId");
+
+create index concurrently if not exists "ix_activityRelations_organizationId_segmentId_include"
+on "activityRelations" ("organizationId", "segmentId")
+include ("platform", "activityId", "memberId");

backend/src/database/migrations/V1747131692__systemSettingsAndSegmentsAgg.sql

@@ -0,0 +1,27 @@
+-- Add "updatedAt" column to track the async aggs updates
+alter table "memberSegmentsAgg"
+    add column if not exists "updatedAt" timestamp with time zone not null default now();
+
+-- For existing rows, set the initial value to the createdAt value
+update "memberSegmentsAgg" set "updatedAt" = "createdAt";
+
+-- Adding "createdAt" and "updatedAt" to make it consistent with the memberSegmentsAgg table.
+alter table "organizationSegmentsAgg"
+    add column if not exists "createdAt" timestamp with time zone not null default now();
+
+alter table "organizationSegmentsAgg"
+    add column if not exists "updatedAt" timestamp with time zone not null default now();
+
+-- table store system wide settings since we are moving away from tenants
+create table if not exists "systemSettings" (
+    name varchar(255) not null primary key,
+    value jsonb not null,
+    description text,
+    "createdAt" timestamp with time zone not null default now(),
+    "updatedAt" timestamp with time zone not null default now()
+);
+
+-- system settings for the display aggs last synced at
+insert into "systemSettings" (name, value) values ('memberDisplayAggsLastSyncedAt', '{"timestamp": "2025-05-13T00:00:00Z"}');
+
+insert into "systemSettings" (name, value) values ('organizationDisplayAggsLastSyncedAt', '{"timestamp": "2025-05-13T00:00:00Z"}');

backend/src/database/migrations/V1747133060__optimizeIndexedEntitiesIndexes.sql

@@ -0,0 +1,3 @@
+-- Index to optimize fetching recently indexed entities
+create index concurrently if not exists "idx_indexed_entities_type_indexed_at_entity_id"
+on "indexed_entities" ("type", "indexed_at", "entity_id");

backend/src/database/repositories/memberRepository.ts

@@ -14,14 +14,12 @@ import {
   Error404,
   Error409,
   RawQueryParser,
-  distinct,
   groupBy,
 } from '@crowd/common'
 import {
   countMembersWithActivities,
   getActiveMembers,
   getLastActivitiesForMembers,
-  getMemberAggregates,
   setMemberDataToActivities,
 } from '@crowd/data-access-layer'
 import { findManyLfxMemberships } from '@crowd/data-access-layer/src/lfx_memberships'
@@ -107,17 +105,6 @@ const { Op } = Sequelize
 
 const log: boolean = false
 
-interface ActivityAggregates {
-  memberId: string
-  segmentId: string
-  activityCount: number
-  activeDaysCount: number
-  lastActive: string
-  activityTypes: string[]
-  activeOn: string[]
-  averageSentiment: number
-}
-
 class MemberRepository {
   static async create(data, options: IRepositoryOptions) {
     if (!data.username && !data.identities) {
@@ -1238,44 +1225,6 @@ class MemberRepository {
     return segments
   }
 
-  static async getActivityAggregates(
-    memberId: string,
-    options: IRepositoryOptions,
-    segmentId?: string,
-  ): Promise<ActivityAggregates> {
-    if (segmentId) {
-      // we load data for a specific segment (can be leaf, parent or grand parent id)
-      const member = (
-        await queryMembersAdvanced(optionsQx(options), options.redis, {
-          filter: { and: [{ id: { eq: memberId } }] },
-          limit: 1,
-          offset: 0,
-          fields: ['activityCount', 'activityTypes', 'activeOn', 'averageSentiment', 'lastActive'],
-          segmentId,
-        })
-      ).rows[0]
-
-      return {
-        activeDaysCount: member?.activeDaysCount || 0,
-        activityCount: member?.activityCount || 0,
-        activityTypes: member?.activityTypes || [],
-        activeOn: member?.activeOn || [],
-        averageSentiment: member?.averageSentiment || 0,
-        lastActive: member?.lastActive || null,
-        memberId,
-        segmentId,
-      }
-    }
-
-    const results = await getMemberAggregates(options.qdb, memberId)
-
-    if (results.length > 0) {
-      return results[0]
-    }
-
-    return null
-  }
-
   static async setAffiliations(
     memberId: string,
     data: MemberSegmentAffiliation[],
@@ -2430,147 +2379,6 @@ class MemberRepository {
     )
   }
 
-  /**
-   * Fill a record with the relations and files (if any)
-   * @param record Record to get relations and files for
-   * @param options IRepository options
-   * @param returnPlain If true: return object, otherwise  return model
-   * @returns The model/object with filled relations and files
-   */
-  static async _populateRelations(
-    record,
-    options: IRepositoryOptions,
-    {
-      returnPlain,
-      segmentId,
-      newIdentities,
-      withActivityAggregates,
-    }: {
-      returnPlain: boolean
-      segmentId: string
-      newIdentities: boolean
-      withActivityAggregates: boolean
-    },
-  ) {
-    if (!record) {
-      return record
-    }
-
-    let output
-
-    if (returnPlain) {
-      output = record.get({ plain: true })
-    } else {
-      output = record
-    }
-
-    const transaction = SequelizeRepository.getTransaction(options)
-
-    if (withActivityAggregates) {
-      const activityAggregates = await MemberRepository.getActivityAggregates(
-        output.id,
-        options,
-        segmentId,
-      )
-
-      output.activeOn = activityAggregates?.activeOn || []
-      output.activityCount = activityAggregates?.activityCount || 0
-      output.activityTypes = activityAggregates?.activityTypes || []
-      output.activeDaysCount = activityAggregates?.activeDaysCount || 0
-      output.averageSentiment = activityAggregates?.averageSentiment || 0
-    }
-
-    output.lastActivity =
-      (
-        await record.getActivities({
-          order: [['timestamp', 'DESC']],
-          limit: 1,
-          transaction,
-        })
-      )[0]?.get({ plain: true }) ?? null
-
-    output.lastActive = output.lastActivity?.timestamp ?? null
-
-    output.numberOfOpenSourceContributions = output.contributions?.length ?? 0
-
-    output.tags = await record.getTags({
-      transaction,
-      order: [['createdAt', 'ASC']],
-      joinTableAttributes: [],
-    })
-
-    output.organizations = await record.getOrganizations({
-      transaction,
-      order: [['createdAt', 'ASC']],
-      joinTableAttributes: ['dateStart', 'dateEnd', 'title', 'source'],
-      through: {
-        where: {
-          deletedAt: null,
-        },
-      },
-    })
-    MemberRepository.sortOrganizations(output.organizations)
-
-    output.noMerge = (
-      await record.getNoMerge({
-        transaction,
-      })
-    ).map((i) => i.id)
-
-    output.toMerge = (
-      await record.getToMerge({
-        transaction,
-      })
-    ).map((i) => i.id)
-
-    const memberIdentities = (await this.getIdentities([record.id], options)).get(record.id)
-
-    if (newIdentities === true) {
-      output.identities = memberIdentities
-      output.verifiedEmails = distinct(
-        memberIdentities
-          .filter((i) => i.verified && i.type === MemberIdentityType.EMAIL)
-          .map((i) => i.value),
-      )
-      output.unverifiedEmails = distinct(
-        memberIdentities
-          .filter((i) => !i.verified && i.type === MemberIdentityType.EMAIL)
-          .map((i) => i.value),
-      )
-      output.verifiedUsernames = distinct(
-        memberIdentities
-          .filter((i) => i.verified && i.type === MemberIdentityType.USERNAME)
-          .map((i) => i.value),
-      )
-      output.unverifiedUsernames = distinct(
-        memberIdentities
-          .filter((i) => !i.verified && i.type === MemberIdentityType.USERNAME)
-          .map((i) => i.value),
-      )
-      output.identityPlatforms = distinct(
-        memberIdentities.filter((i) => i.verified).map((i) => i.platform),
-      )
-    } else {
-      output.username = {}
-
-      for (const identity of memberIdentities.filter(
-        (i) => i.type === MemberIdentityType.USERNAME,
-      )) {
-        if (output.username[identity.platform]) {
-          output.username[identity.platform].push(identity.value)
-        } else {
-          output.username[identity.platform] = [identity.value]
-        }
-      }
-
-      output.identities = Object.keys(output.username)
-    }
-
-    output.affiliations = await this.getAffiliations(record.id, options)
-
-    return output
-  }
-
   static async updateMemberOrganizations(
     record,
     organizations,