chore(integrations): added segmentId to filters for populate activity relations workflow

Signed-off-by: Uroš Marolt <uros@marolt.me>

Author: themarolt

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

@@ -28,11 +28,16 @@ export async function markActivitiesAsIndexed(activitiesRedisKey: string): Promi
   return lastSyncedTimestamp
 }
 
-export async function getActivitiesToCopy(latestSyncedActivityTimestamp: string, limit: number) {
+export async function getActivitiesToCopy(
+  latestSyncedActivityTimestamp: string,
+  limit: number,
+  segmentIds?: string[],
+) {
   const activities = await getActivityRelationsSortedByTimestamp(
     svc.questdbSQL,
     latestSyncedActivityTimestamp,
     limit,
+    segmentIds,
   )
 
   if (activities.length === 0) {

services/apps/script_executor_worker/src/types.ts

@@ -27,6 +27,7 @@ export interface IPopulateActivityRelationsArgs {
   batchSizePerRun: number
   deleteIndexedEntities?: boolean
   latestSyncedActivityTimestamp?: string
+  segmentIds?: string[]
 }
 
 export interface IScriptBatchTestArgs {

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

@@ -25,6 +25,7 @@ export async function populateActivityRelations(
   let { activitiesLength, activitiesRedisKey, lastTimestamp } = await activity.getActivitiesToCopy(
     latestSyncedActivityTimestamp ?? undefined,
     BATCH_SIZE_PER_RUN,
+    args.segmentIds,
   )
 
   if (activitiesLength === 0) {
@@ -44,6 +45,7 @@ export async function populateActivityRelations(
       const result = await activity.getActivitiesToCopy(
         lastTimestamp,
         BATCH_SIZE_PER_RUN * batchSizeMultiplier,
+        args.segmentIds,
       )
       activitiesLength = result.activitiesLength
       activitiesRedisKey = result.activitiesRedisKey

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

@@ -1715,11 +1715,16 @@ export async function getActivityRelationsSortedByTimestamp(
   qdbConn: DbConnOrTx,
   cursorActivityTimestamp?: string,
   limit = 100,
+  segmentIds?: string[],
 ): Promise<IActivityRelationsCreateData[]> {
-  let cursorQuery = ''
+  const conditions: string[] = [`"deletedAt" is null`]
 
   if (cursorActivityTimestamp) {
-    cursorQuery = `AND "timestamp" >= $(cursorActivityTimestamp)`
+    conditions.push('timestamp >= $(cursorActivityTimestamp)')
+  }
+
+  if (segmentIds && segmentIds.length > 0) {
+    conditions.push('"segmentId" in ($(segmentIds:csv))')
   }
 
   const query = `
@@ -1737,14 +1742,14 @@ export async function getActivityRelationsSortedByTimestamp(
       username,
       "objectMemberUsername"
     FROM activities
-    WHERE "deletedAt" IS NULL
-    ${cursorQuery}
+    WHERE ${conditions.join(' AND ')}
     ORDER BY "timestamp" asc
     LIMIT ${limit}
   `
 
   const rows = await qdbConn.any(query, {
     cursorActivityTimestamp,
+    segmentIds,
     limit,
   })