improve pagination by using timestamp instead

Author: epipav

services/apps/script_executor_worker/src/activities/populate-activity-relations/index.ts

@@ -1,5 +1,9 @@
 import { getDefaultTenantId } from '@crowd/common'
-import { createOrUpdateRelations, getActivitiesSortedById } from '@crowd/data-access-layer'
+import {
+  createOrUpdateRelations,
+  getActivityRelationsSortedByTimestamp,
+  getActivityTimestampById,
+} from '@crowd/data-access-layer'
 import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor'
 import { IndexedEntityType, IndexingRepository } from '@crowd/opensearch'
 
@@ -26,7 +30,19 @@ export async function markActivitiesAsIndexed(activityIds: string[]): Promise<vo
 }
 
 export async function getActivitiesToCopy(latestSyncedActivityId: string, limit: number) {
-  const activities = await getActivitiesSortedById(svc.questdbSQL, latestSyncedActivityId, limit)
+  let latestSyncedActivityTimestamp = undefined
+  if (latestSyncedActivityId) {
+    latestSyncedActivityTimestamp = await getActivityTimestampById(
+      svc.questdbSQL,
+      latestSyncedActivityId,
+    )
+  }
+
+  const activities = await getActivityRelationsSortedByTimestamp(
+    svc.questdbSQL,
+    latestSyncedActivityTimestamp ? latestSyncedActivityTimestamp.timestamp : undefined,
+    limit,
+  )
   return activities
 }
 

services/apps/script_executor_worker/src/types.ts

@@ -30,4 +30,5 @@ export interface IFixOrgIdentitiesWithWrongUrlsArgs {
 export interface IPopulateActivityRelationsArgs {
   batchSizePerRun: number
   deleteIndexedEntities?: boolean
+  lastIndexedActivityId?: string
 }

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

@@ -17,7 +17,8 @@ export async function populateActivityRelations(
     await activity.deleteActivityIdsFromIndexedEntities()
   }
 
-  const latestSyncedActivityId = await activity.getLatestSyncedActivityId()
+  const latestSyncedActivityId =
+    args.lastIndexedActivityId || (await activity.getLatestSyncedActivityId())
 
   const activitiesToCopy = await activity.getActivitiesToCopy(
     latestSyncedActivityId ?? undefined,
@@ -28,11 +29,19 @@ export async function populateActivityRelations(
     return
   }
 
+  if (activitiesToCopy.length < BATCH_SIZE_PER_RUN) {
+    const lastSyncedActivityId = await activity.getLatestSyncedActivityId()
+    if (lastSyncedActivityId === args.lastIndexedActivityId)  {
+      return
+    }
+  }
+
   await activity.createRelations(activitiesToCopy)
 
   await activity.markActivitiesAsIndexed(activitiesToCopy.map((a) => a.id))
 
   await continueAsNew<typeof populateActivityRelations>({
     batchSizePerRun: args.batchSizePerRun,
+    lastIndexedActivityId: activitiesToCopy[activitiesToCopy.length - 1].id,
   })
 }

services/libs/data-access-layer/src/activities/sql.ts

@@ -10,8 +10,8 @@ import {
   ActivityDisplayVariant,
   IActivityBySentimentMoodResult,
   IActivityByTypeAndPlatformResult,
-  IActivityCreateData,
   IActivityData,
+  IActivityDbBase,
   IEnrichableMemberIdentityActivityAggregate,
   IMemberIdentity,
   ITimeseriesDatapoint,
@@ -75,6 +75,30 @@ export async function getActivitiesById(
   return data.rows
 }
 
+/**
+ * Finds activity timestamp by id, without tenant or segment filters
+ * @param qdbConn
+ * @param id
+ * @returns IActivityCreateData
+ */
+export async function getActivityTimestampById(
+  qdbConn: DbConnOrTx,
+  id: string,
+): Promise<Partial<IActivityDbBase>> {
+  const query = `
+    SELECT timestamp
+    FROM activities
+    WHERE "deletedAt" IS NULL
+    and id = $(id)
+  `
+
+  const rows = await qdbConn.any(query, {
+    id,
+  })
+
+  return rows.length > 0 ? rows[0] : null
+}
+
 const ACTIVITY_UPDATABLE_COLUMNS: ActivityColumn[] = [
   'type',
   'isContribution',
@@ -1714,28 +1738,38 @@ export async function moveActivityRelationsToAnotherOrganization(
   } while (rowsUpdated === batchSize)
 }
 
-export async function getActivitiesSortedById(
+export async function getActivityRelationsSortedByTimestamp(
   qdbConn: DbConnOrTx,
-  cursorActivityId?: string,
+  cursorActivityTimestamp?: string,
   limit = 100,
 ) {
   let cursorQuery = ''
 
-  if (cursorActivityId) {
-    cursorQuery = `AND id > $(cursorActivityId)::uuid`
+  if (cursorActivityTimestamp) {
+    cursorQuery = `AND timestamp >= $(cursorActivityTimestamp)`
   }
 
   const query = `
-    SELECT *
+    SELECT 
+      id,
+      "memberId",
+      "objectMemberId",
+      "organizationId",
+      "conversationId",
+      "parentId",
+      "segmentId",
+      platform,
+      username,
+      "objectMemberUsername"
     FROM activities
     WHERE "deletedAt" IS NULL
     ${cursorQuery}
-    ORDER BY id::text asc
+    ORDER BY timestamp asc
     LIMIT ${limit}
   `
 
   const rows = await qdbConn.any(query, {
-    cursorActivityId,
+    cursorActivityTimestamp,
     limit,
   })
 

services/libs/opensearch/src/repo/indexing.repo.ts

@@ -38,7 +38,7 @@ export class IndexingRepository extends RepositoryBase<IndexingRepository> {
       select entity_id
       from indexed_entities
       where type = $(type)
-      order by entity_id desc
+      order by indexed_at desc
       limit 1
       `,
       {

services/libs/types/src/activities.ts

@@ -1,6 +1,30 @@
 import { PlatformType } from './enums/platforms'
 import { IMemberData } from './members'
 
+
+export interface IActivityDbBase {
+  id: string
+  type: string
+  isContribution: boolean
+  score: number
+  timestamp: string
+  platform: string
+  sourceId: string
+  sourceParentId?: string
+  memberId: string
+  username: string
+  objectMemberId?: string
+  objectMemberUsername?: string
+  attributes: Record<string, unknown>
+  body?: string
+  title?: string
+  channel?: string
+  url?: string
+  tenantId?: string
+  organizationId?: string
+  segmentId?: string
+}
+
 export interface IActivityCreateData {
   id?: string
   type: string