Handle activity relations without username or platform or segmentId (#2863)

Author: epipav

services/apps/script_executor_worker/src/activities.ts

@@ -24,7 +24,7 @@ import {
 import {
   createRelations,
   getActivitiesToCopy,
-  getLatestSyncedActivityCreatedAt,
+  getLatestSyncedActivityTimestamp,
   markActivitiesAsIndexed,
   resetIndexedIdentities,
 } from './activities/populate-activity-relations'
@@ -54,7 +54,7 @@ export {
   createRelations,
   resetIndexedIdentities,
   getActivitiesToCopy,
-  getLatestSyncedActivityCreatedAt,
+  getLatestSyncedActivityTimestamp,
   markActivitiesAsIndexed,
   syncMembersBatch,
   getMembersForSync,

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

@@ -1,6 +1,6 @@
 import {
   createOrUpdateRelations,
-  getActivityRelationsSortedByCreatedAt,
+  getActivityRelationsSortedByTimestamp,
 } from '@crowd/data-access-layer'
 import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor'
 import { RedisCache } from '@crowd/redis'
@@ -9,27 +9,27 @@ import { svc } from '../../main'
 
 export async function resetIndexedIdentities(): Promise<void> {
   const redisCache = new RedisCache(`activity-relations-data`, svc.redis, svc.log)
-  await redisCache.delete('latest-synced-activity-created-at')
+  await redisCache.delete('last-synced-activity-timestamp')
 }
 
-export async function getLatestSyncedActivityCreatedAt(): Promise<string> {
+export async function getLatestSyncedActivityTimestamp(): Promise<string> {
   const redisCache = new RedisCache(`activity-relations-data`, svc.redis, svc.log)
-  const result = await redisCache.get('latest-synced-activity-created-at')
+  const result = await redisCache.get('last-synced-activity-timestamp')
   return result || null
 }
 
 export async function markActivitiesAsIndexed(activitiesRedisKey: string): Promise<string> {
   const activities = await getActivitiyDataFromRedis(activitiesRedisKey)
   const redisCache = new RedisCache(`activity-relations-data`, svc.redis, svc.log)
-  const lastSyncedCreatedAt = activities[activities.length - 1].createdAt
-  await redisCache.set('latest-synced-activity-created-at', lastSyncedCreatedAt)
-  return lastSyncedCreatedAt
+  const lastSyncedTimestamp = activities[activities.length - 1].timestamp
+  await redisCache.set('last-synced-activity-timestamp', lastSyncedTimestamp)
+  return lastSyncedTimestamp
 }
 
-export async function getActivitiesToCopy(latestSyncedActivityCreatedAt: string, limit: number) {
-  const activities = await getActivityRelationsSortedByCreatedAt(
+export async function getActivitiesToCopy(latestSyncedActivityTimestamp: string, limit: number) {
+  const activities = await getActivityRelationsSortedByTimestamp(
     svc.questdbSQL,
-    latestSyncedActivityCreatedAt,
+    latestSyncedActivityTimestamp,
     limit,
   )
 
@@ -44,7 +44,7 @@ export async function getActivitiesToCopy(latestSyncedActivityCreatedAt: string,
   return {
     activitiesRedisKey: key,
     activitiesLength: activities.length,
-    lastCreatedAt: activities[activities.length - 1].createdAt,
+    lastTimestamp: activities[activities.length - 1].timestamp,
   }
 }
 
@@ -70,7 +70,7 @@ export async function createRelations(activitiesRedisKey): Promise<void> {
 
 export async function saveActivityDataToRedis(key: string, activities): Promise<void> {
   const redisCache = new RedisCache(`activity-relations-data`, svc.redis, svc.log)
-  await redisCache.set(key, JSON.stringify(activities), 30)
+  await redisCache.set(key, JSON.stringify(activities), 360)
 }
 
 export async function getActivitiyDataFromRedis(key: string) {

services/apps/script_executor_worker/src/schedules/schedulePopulateActivityRelations.ts

@@ -25,7 +25,7 @@ export const schedulePopulateActivityRelations = async () => {
         },
         args: [
           {
-            batchSizePerRun: 10000,
+            batchSizePerRun: 50000,
             deleteIndexedEntities: false,
           },
         ],

services/apps/script_executor_worker/src/types.ts

@@ -23,7 +23,7 @@ export interface IFixOrgIdentitiesWithWrongUrlsArgs {
 export interface IPopulateActivityRelationsArgs {
   batchSizePerRun: number
   deleteIndexedEntities?: boolean
-  latestSyncedActivityCreatedAt?: string
+  latestSyncedActivityTimestamp?: string
 }
 export interface ISyncMembersArgs {
   batchSize?: number

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

@@ -12,19 +12,19 @@ export async function populateActivityRelations(
   args: IPopulateActivityRelationsArgs,
 ): Promise<void> {
   const BATCH_SIZE_PER_RUN = args.batchSizePerRun || 1000
-  let latestSyncedActivityCreatedAt
+  let latestSyncedActivityTimestamp
 
   if (args.deleteIndexedEntities) {
     await activity.resetIndexedIdentities()
-    latestSyncedActivityCreatedAt = null
+    latestSyncedActivityTimestamp = null
   } else {
-    latestSyncedActivityCreatedAt =
-      args.latestSyncedActivityCreatedAt || (await activity.getLatestSyncedActivityCreatedAt())
+    latestSyncedActivityTimestamp =
+      args.latestSyncedActivityTimestamp || (await activity.getLatestSyncedActivityTimestamp())
   }
 
-  const { activitiesLength, activitiesRedisKey, lastCreatedAt } =
+  const { activitiesLength, activitiesRedisKey, lastTimestamp } =
     await activity.getActivitiesToCopy(
-      latestSyncedActivityCreatedAt ?? undefined,
+      latestSyncedActivityTimestamp ?? undefined,
       BATCH_SIZE_PER_RUN,
     )
 
@@ -33,7 +33,7 @@ export async function populateActivityRelations(
   }
 
   if (activitiesLength < BATCH_SIZE_PER_RUN) {
-    if (lastCreatedAt === args.latestSyncedActivityCreatedAt) {
+    if (lastTimestamp === args.latestSyncedActivityTimestamp) {
       return
     }
   }
@@ -44,6 +44,6 @@ export async function populateActivityRelations(
 
   await continueAsNew<typeof populateActivityRelations>({
     batchSizePerRun: args.batchSizePerRun,
-    latestSyncedActivityCreatedAt: lastCreatedAt,
+    latestSyncedActivityTimestamp: lastTimestamp,
   })
 }

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

@@ -1423,6 +1423,18 @@ export async function createOrUpdateRelations(
   qe: QueryExecutor,
   data: IActivityRelationCreateOrUpdateData,
 ): Promise<void> {
+  if (data.username === undefined || data.username === null) {
+    return
+  }
+
+  if (data.platform === undefined || data.platform === null) {
+    return
+  }
+
+  if (data.segmentId === undefined || data.segmentId === null) {
+    return
+  }
+
   // check objectMember exists
   if (data.objectMemberId !== undefined && data.objectMemberId !== null) {
     let objectMember = await qe.select(
@@ -1729,21 +1741,22 @@ export async function moveActivityRelationsToAnotherOrganization(
   } while (rowsUpdated === batchSize)
 }
 
-export async function getActivityRelationsSortedByCreatedAt(
+export async function getActivityRelationsSortedByTimestamp(
   qdbConn: DbConnOrTx,
-  cursorActivityCreatedAt?: string,
+  cursorActivityTimestamp?: string,
   limit = 100,
 ) {
   let cursorQuery = ''
 
-  if (cursorActivityCreatedAt) {
-    cursorQuery = `AND "createdAt" >= $(cursorActivityCreatedAt)`
+  if (cursorActivityTimestamp) {
+    cursorQuery = `AND "timestamp" >= $(cursorActivityTimestamp)`
   }
 
   const query = `
     SELECT 
       id,
       "memberId",
+      timestamp,
       "createdAt",
       "objectMemberId",
       "organizationId",
@@ -1756,12 +1769,12 @@ export async function getActivityRelationsSortedByCreatedAt(
     FROM activities
     WHERE "deletedAt" IS NULL
     ${cursorQuery}
-    ORDER BY "createdAt" asc
+    ORDER BY "timestamp" asc
     LIMIT ${limit}
   `
 
   const rows = await qdbConn.any(query, {
-    cursorActivityCreatedAt,
+    cursorActivityTimestamp,
     limit,
   })