use different approach to paginate

Author: skwowet

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

@@ -267,7 +267,7 @@ export async function updateMemberMergeSuggestionsLastGeneratedAt(tenantId: stri
 export async function getMembers(
   tenantId: string,
   batchSize: number,
-  afterMemberId?: string,
+  // afterMemberId?: string,
   lastGeneratedAt?: string,
   exactMemberId?: string,
 ): Promise<IMemberBaseForMergeSuggestions[]> {
@@ -278,7 +278,7 @@ export async function getMembers(
         and: [
           { [MemberField.TENANT_ID]: { eq: tenantId } },
           exactMemberId ? { [MemberField.ID]: { eq: exactMemberId } } : null,
-          afterMemberId ? { [MemberField.ID]: { gt: afterMemberId } } : null,
+          // afterMemberId ? { [MemberField.ID]: { gt: afterMemberId } } : null,
           lastGeneratedAt ? { [MemberField.CREATED_AT]: { gt: lastGeneratedAt } } : null,
         ],
       },
@@ -287,6 +287,7 @@ export async function getMembers(
         MemberField.TENANT_ID,
         MemberField.DISPLAY_NAME,
         MemberField.ATTRIBUTES,
+        MemberField.CREATED_AT,
       ],
       limit: batchSize,
     })

services/apps/merge_suggestions_worker/src/types.ts

@@ -70,7 +70,8 @@ export interface ILLMBody {
 
 export interface IProcessGenerateMemberMergeSuggestionsArgs {
   tenantId: string
-  lastUuid?: string
+  lastGeneratedAt?: string
+  testRun?: number
 }
 
 export interface IProcessGenerateOrganizationMergeSuggestionsArgs {

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

@@ -19,15 +19,18 @@ export async function generateMemberMergeSuggestions(
   const PARALLEL_SUGGESTION_PROCESSING = 10
   const SIMILARITY_CONFIDENCE_SCORE_THRESHOLD = 0.75
 
-  let lastUuid: string = args.lastUuid || null
+  let lastGeneratedAt: string = args.lastGeneratedAt || null
 
   // 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)
+  if (!lastGeneratedAt) {
+    lastGeneratedAt = await activity.findTenantsLatestMemberSuggestionGeneratedAt(args.tenantId)
+  }
+
+  console.log('[generateMemberMergeSuggestions] Last generated at:', lastGeneratedAt)
 
   const result: IMemberBaseForMergeSuggestions[] = await activity.getMembers(
     args.tenantId,
     PAGE_SIZE,
-    lastUuid,
     lastGeneratedAt,
   )
 
@@ -51,7 +54,9 @@ export async function generateMemberMergeSuggestions(
     return
   }
 
-  lastUuid = result.length > 0 ? result[result.length - 1]?.id : null
+  const lastProcessedAt = result.length > 0 ? result[result.length - 1]?.createdAt : null
+
+  console.log('[generateMemberMergeSuggestions] Last processed member createdAt:', lastProcessedAt)
 
   const allMergeSuggestions: IMemberMergeSuggestion[] = []
 
@@ -109,5 +114,15 @@ export async function generateMemberMergeSuggestions(
     )
   }
 
-  await continueAsNew<typeof generateMemberMergeSuggestions>({ tenantId: args.tenantId, lastUuid })
+  // if testRun > 3, we'll stop the workflow
+  // todo: remove this
+  if (args.testRun > 3) {
+    return
+  }
+
+  await continueAsNew<typeof generateMemberMergeSuggestions>({
+    tenantId: args.tenantId,
+    lastGeneratedAt: lastProcessedAt,
+    testRun: args.testRun + 1,
+  })
 }

services/libs/types/src/members.ts

@@ -166,6 +166,7 @@ export interface IMemberBaseForMergeSuggestions {
   id: string
   displayName: string
   attributes: IAttributes
+  createdAt: string
 }
 
 export interface IMemberWithAggregatesForMergeSuggestions extends IMemberBaseForMergeSuggestions {