use lastTimestamp

Author: skwowet

services/apps/script_executor_worker/src/activities/fix-member-affilations/index.ts

@@ -6,10 +6,10 @@ import { svc } from '../../main'
 export async function getSegmentMembers(
   segmentId: string,
   limit: number,
-  offset: number,
-): Promise<string[]> {
+  lastTimestamp?: string,
+): Promise<{ memberIds: string[]; lastTimestamp: string }> {
   const memberRepo = new MemberRepository(svc.postgres.reader.connection(), svc.log, svc.questdbSQL)
-  return memberRepo.getSegmentMembers(segmentId, limit, offset)
+  return memberRepo.getSegmentMembers(segmentId, limit, lastTimestamp)
 }
 
 export async function calculateMemberAffiliations(memberId: string): Promise<void> {

services/apps/script_executor_worker/src/types.ts

@@ -36,6 +36,6 @@ export interface ISyncMembersArgs {
 export interface IFixMemberAffilationsArgs {
   segmentId: string
   limit?: number
-  offset?: number
+  lastTimestamp?: string
   testRun?: boolean
 }

services/apps/script_executor_worker/src/workflows/fixMemberAffilations.ts

@@ -16,13 +16,19 @@ const syncActivity = proxyActivities<typeof syncActivities>({
 
 export async function fixMemberAffilations(args: IFixMemberAffilationsArgs): Promise<void> {
   const limit = args.limit ?? 200
-  const offset = args.offset ?? 0
+  const lastTimestamp = args.lastTimestamp
 
   console.log(
-    `Fixing member affiliations for segment ${args.segmentId} with limit ${limit} and offset ${offset}`,
+    `Fixing member affiliations for segment ${args.segmentId} with limit ${limit}${
+      lastTimestamp ? ` and lastTimestamp ${lastTimestamp}` : ''
+    }`,
   )
 
-  const memberIds = await activity.getSegmentMembers(args.segmentId, limit, offset)
+  const { memberIds, lastTimestamp: newLastTimestamp } = await activity.getSegmentMembers(
+    args.segmentId,
+    limit,
+    lastTimestamp,
+  )
 
   if (memberIds.length === 0) {
     console.log('No more members to fix!')
@@ -43,6 +49,6 @@ export async function fixMemberAffilations(args: IFixMemberAffilationsArgs): Pro
 
   await continueAsNew<typeof fixMemberAffilations>({
     ...args,
-    offset: offset + limit,
+    lastTimestamp: newLastTimestamp,
   })
 }

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

@@ -173,24 +173,33 @@ class MemberRepository {
     return results.map((r) => r.id)
   }
 
-  public async getSegmentMembers(segmentId: string, limit = 200, offset = 0): Promise<string[]> {
+  public async getSegmentMembers(
+    segmentId: string,
+    limit = 200,
+    lastTimestamp?: string,
+  ): Promise<{ memberIds: string[]; lastTimestamp: string }> {
+    const timestampFilter = lastTimestamp ? `AND "timestamp" < '${lastTimestamp}'` : ''
+
     const results = await this.questdbSQL.query(
       `
-      SELECT DISTINCT "memberId"
-      FROM activities
-      WHERE "segmentId" = $(segmentId)
-      AND "deletedAt" IS NULL
-      ORDER BY "timestamp" DESC
-      LIMIT $(limit), $(offset)
+      WITH distinct_members AS (
+        SELECT DISTINCT "memberId", max("timestamp") as last_activity
+        FROM activities
+        WHERE "segmentId" = '${segmentId}'
+        ${timestampFilter}
+        GROUP BY "memberId"
+      )
+      SELECT "memberId", last_activity as "lastTimestamp"
+      FROM distinct_members
+      ORDER BY last_activity DESC
+      LIMIT ${limit}
       `,
-      {
-        segmentId,
-        limit,
-        offset,
-      },
     )
 
-    return results.map((r) => r.memberId)
+    return {
+      memberIds: results.map((r) => r.memberId),
+      lastTimestamp: results.length > 0 ? results[results.length - 1].lastTimestamp : undefined,
+    }
   }
 }