suggestions use memberMergeSuggestionsLastGeneratedAt instead

backend/src/database/migrations/U1704395824__memberMergeSuggestionsLastGeneratedAt.sql

@@ -0,0 +1 @@
+alter table "tenants" drop column "memberMergeSuggestionsLastGeneratedAt";

backend/src/database/migrations/V1704395824__memberMergeSuggestionsLastGeneratedAt.sql

@@ -0,0 +1,2 @@
+alter table "tenants"
+add column "memberMergeSuggestionsLastGeneratedAt" timestamp with time zone null;

services/apps/merge_suggestions_worker/src/activities.ts

@@ -3,13 +3,15 @@ import {
   getMergeSuggestions,
   addToMerge,
   getMembers,
-  findTenantsLatestSuggestionCreatedAt,
+  findTenantsLatestMemberSuggestionGeneratedAt,
+  updateMemberMergeSuggestionsLastGeneratedAt,
 } from './activities/member-merge-suggestions/getMergeSuggestions'
 
 export {
   getAllTenants,
   getMergeSuggestions,
   getMembers,
   addToMerge,
-  findTenantsLatestSuggestionCreatedAt,
+  findTenantsLatestMemberSuggestionGeneratedAt,
+  updateMemberMergeSuggestionsLastGeneratedAt,
 }

services/apps/merge_suggestions_worker/src/activities/member-merge-suggestions/getMergeSuggestions.ts

@@ -193,19 +193,29 @@ export async function addToMerge(suggestions: IMemberMergeSuggestion[]): Promise
   await memberMergeSuggestionsRepo.addToMerge(suggestions)
 }
 
-export async function findTenantsLatestSuggestionCreatedAt(tenantId: string): Promise<string> {
+export async function findTenantsLatestMemberSuggestionGeneratedAt(
+  tenantId: string,
+): Promise<string> {
+  const memberMergeSuggestionsRepo = new MemberMergeSuggestionsRepository(
+    svc.postgres.writer.connection(),
+    svc.log,
+  )
+  return memberMergeSuggestionsRepo.findTenantsLatestMemberSuggestionGeneratedAt(tenantId)
+}
+
+export async function updateMemberMergeSuggestionsLastGeneratedAt(tenantId: string): Promise<void> {
   const memberMergeSuggestionsRepo = new MemberMergeSuggestionsRepository(
     svc.postgres.writer.connection(),
     svc.log,
   )
-  return memberMergeSuggestionsRepo.findTenantsLatestSuggestionCreatedAt(tenantId)
+  await memberMergeSuggestionsRepo.updateMemberMergeSuggestionsLastGeneratedAt(tenantId)
 }
 
 export async function getMembers(
   tenantId: string,
   batchSize: number = 100,
   afterMemberId?: string,
-  lastCreatedAt?: string,
+  lastGeneratedAt?: string,
 ): Promise<IMemberPartialAggregatesOpensearch[]> {
   try {
     const queryBody: IMemberQueryBody = {
@@ -248,11 +258,11 @@ export async function getMembers(
       })
     }
 
-    if (lastCreatedAt) {
+    if (lastGeneratedAt) {
       queryBody.query.bool.filter.push({
         range: {
           date_createdAt: {
-            gt: new Date(lastCreatedAt).toISOString(),
+            gt: new Date(lastGeneratedAt).toISOString(),
           },
         },
       })

services/apps/merge_suggestions_worker/src/repo/memberMergeSuggestions.repo.ts

@@ -1,7 +1,7 @@
 import { DbConnection, DbTransaction } from '@crowd/database'
 import { Logger } from '@crowd/logging'
 import { IMemberMergeSuggestion } from '@crowd/types'
-import { IMemberId, IMemberNoMerge, IMemberMergeSuggestionsLatestCreatedAt } from 'types'
+import { IMemberId, IMemberMergeSuggestionsLatestGeneratedAt, IMemberNoMerge } from 'types'
 import { chunkArray, removeDuplicateSuggestions } from 'utils'
 
 class MemberMergeSuggestionsRepository {
@@ -34,19 +34,34 @@ class MemberMergeSuggestionsRepository {
     }
   }
 
-  async findTenantsLatestSuggestionCreatedAt(tenantId: string): Promise<string> {
+  async findTenantsLatestMemberSuggestionGeneratedAt(tenantId: string): Promise<string> {
     try {
-      const result: IMemberMergeSuggestionsLatestCreatedAt = await this.connection.oneOrNone(
+      const result: IMemberMergeSuggestionsLatestGeneratedAt = await this.connection.oneOrNone(
         `
-        select max(mtm."createdAt") as "latestCreatedAt"
-        from "memberToMerge" mtm
-        inner join members m on mtm."memberId" = m.id
-        where m."tenantId" = $(tenantId);`,
+        select "memberMergeSuggestionsLastGeneratedAt"
+        from tenants
+        where "id" = $(tenantId);`,
+        {
+          tenantId,
+        },
+      )
+      return result?.memberMergeSuggestionsLastGeneratedAt
+    } catch (err) {
+      throw new Error(err)
+    }
+  }
+
+  async updateMemberMergeSuggestionsLastGeneratedAt(tenantId: string): Promise<void> {
+    try {
+      await this.connection.any(
+        `
+          update tenants set "memberMergeSuggestionsLastGeneratedAt" = now()
+          where "id" = $(tenantId);
+        `,
         {
           tenantId,
         },
       )
-      return result?.latestCreatedAt
     } catch (err) {
       throw new Error(err)
     }

services/apps/merge_suggestions_worker/src/types.ts

@@ -79,8 +79,8 @@ export interface IMemberNoMerge {
   noMergeId: string
 }
 
-export interface IMemberMergeSuggestionsLatestCreatedAt {
-  latestCreatedAt: string
+export interface IMemberMergeSuggestionsLatestGeneratedAt {
+  memberMergeSuggestionsLastGeneratedAt: string
 }
 
 export interface IMemberId {

services/apps/merge_suggestions_worker/src/workflows/generateMemberMergeSuggestions.ts

@@ -16,11 +16,11 @@ export async function generateMemberMergeSuggestions(
   let result: IMemberPartialAggregatesOpensearch[]
   let lastUuid: string
 
-  // get the latest createdAt of tenant's member suggestions, we'll only get members created after that for new suggestions
-  const lastCreatedAt = await activity.findTenantsLatestSuggestionCreatedAt(args.tenantId)
+  // get the latest generation time of tenant's member suggestions, we'll only get members created after that for new suggestions
+  const lastGeneratedAt = await activity.findTenantsLatestMemberSuggestionGeneratedAt(args.tenantId)
 
   do {
-    result = await activity.getMembers(args.tenantId, PAGE_SIZE, lastUuid, lastCreatedAt)
+    result = await activity.getMembers(args.tenantId, PAGE_SIZE, lastUuid, lastGeneratedAt)
 
     lastUuid = result.length > 0 ? result[result.length - 1]?.uuid_memberId : null
 
@@ -46,4 +46,6 @@ export async function generateMemberMergeSuggestions(
       await activity.addToMerge(allMergeSuggestions)
     }
   } while (result.length > 0)
+
+  await activity.updateMemberMergeSuggestionsLastGeneratedAt(args.tenantId)
 }