chunking inserts 1k each

Author: epipav

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

@@ -1,4 +1,6 @@
+import { chunkArray } from '@crowd/common'
 import {
+  IActivityRelationsCreateData,
   createOrUpdateRelations,
   getActivityRelationsSortedByTimestamp,
 } from '@crowd/data-access-layer'
@@ -49,23 +51,30 @@ export async function getActivitiesToCopy(latestSyncedActivityTimestamp: string,
 }
 
 export async function createRelations(activitiesRedisKey): Promise<void> {
-  const activities = await getActivitiyDataFromRedis(activitiesRedisKey)
-  const promises = activities.map(async (activity) =>
-    createOrUpdateRelations(pgpQx(svc.postgres.writer.connection()), {
-      activityId: activity.id,
-      memberId: activity.memberId,
-      platform: activity.platform,
-      segmentId: activity.segmentId,
-      username: activity.username,
-      conversationId: activity.conversationId,
-      objectMemberId: activity.objectMemberId,
-      objectMemberUsername: activity.objectMemberUsername,
-      organizationId: activity.organizationId,
-      parentId: activity.parentId,
-    }),
-  )
+  const activities: IActivityRelationsCreateData[] =
+    await getActivitiyDataFromRedis(activitiesRedisKey)
+
+  const chunkSize = 1000
+  const activityChunks = chunkArray(activities, chunkSize)
 
-  await Promise.all(promises)
+  for (const chunk of activityChunks) {
+    const promises = chunk.map((activity) =>
+      createOrUpdateRelations(pgpQx(svc.postgres.writer.connection()), {
+        activityId: activity.id,
+        memberId: activity.memberId,
+        platform: activity.platform,
+        segmentId: activity.segmentId,
+        username: activity.username,
+        conversationId: activity.conversationId,
+        objectMemberId: activity.objectMemberId,
+        objectMemberUsername: activity.objectMemberUsername,
+        organizationId: activity.organizationId,
+        parentId: activity.parentId,
+      }),
+    )
+
+    await Promise.all(promises)
+  }
 }
 
 export async function saveActivityDataToRedis(key: string, activities): Promise<void> {

services/libs/common/src/array.ts

@@ -134,3 +134,11 @@ export const hasIntersection = (arr1: string[], arr2: string[]): boolean => {
   const set1 = new Set(arr1)
   return arr2.some((item) => set1.has(item))
 }
+
+export function chunkArray<T>(array: T[], chunkSize: number): T[][] {
+  const chunks = []
+  for (let i = 0; i < array.length; i += chunkSize) {
+    chunks.push(array.slice(i, i + chunkSize))
+  }
+  return chunks
+}

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

@@ -1741,11 +1741,26 @@ export async function moveActivityRelationsToAnotherOrganization(
   } while (rowsUpdated === batchSize)
 }
 
+export interface IActivityRelationsCreateData {
+  id: string
+  memberId: string
+  timestamp: string
+  createdAt: string
+  objectMemberId?: string
+  organizationId?: string
+  conversationId?: string
+  parentId?: string
+  segmentId: string
+  platform: string
+  username: string
+  objectMemberUsername?: string
+}
+
 export async function getActivityRelationsSortedByTimestamp(
   qdbConn: DbConnOrTx,
   cursorActivityTimestamp?: string,
   limit = 100,
-) {
+): Promise<IActivityRelationsCreateData[]> {
   let cursorQuery = ''
 
   if (cursorActivityTimestamp) {